ServiceGroup should support dynamically adding Services while its running

[fabianfett]

There are scenarios in which users need a running Service in order to retrieve data. This data shall then be used to spawn another Service.

An example for this could be querying AWS SSM for a secret that is then used inside a Lambda.

I imagine the code to look something like this:

import Soto
import SotoSSM

enum MyLambda {
    static func main() async throws {
        let awsClient = AWSClient()
        let ssmClient = SSMClient(awsClient)

        try await withTaskGroup { taskGroup in
            let serviceGroup = ServiceGroup([awsClient])
            taskGroup.addTask {
                try await serviceGroup.run()
            }

            let mySecret = try await ssmClient.getSecret()

            let runtime = LambdaRuntime { (event: Request, context) in
                // use mySecret
            }

            // add runtime to already running serviceGroup.
            serviceGroup.add(runtime)
        }
    }
}

[FranzBusch]

I think we can totally do that and be able to add dynamic services to the group. I was briefly thinking if having dependency between services during startup could be used here. I think it could but it would become awkward to spell. So yeah great feature requests!

[sliemeobn]

I was about to give this a go, but got stuck on a fundamental API question:

The inner plumbings are quite straight forward I would say (some async sequence of ServiceConfiguration to pipe new services into the main run loop and set them up there.)

I was going in with a non-throwing nonisolated public func addService(...), but that would essentially mean: no access to the group's state, no feedback whatsoever if the group was even added or its run called.

public actor ServiceGroup {
    //...
    private let 
    nonisolated private let addedServices: AsyncStream<ServiceGroupConfiguration.ServiceConfiguration>.Continuation

    //...
    nonisolated public func addService(_ serviceConfiguration: ServiceGroupConfiguration.ServiceConfiguration) {
        // Without making `addService` throwing and/or async, there is not a lot we could do with the result of yield
        addedServices.yield(serviceConfiguration)
    }

And, you'd theoretically have an unbounded buffer of added services if nobody ever calls run.

The alternative is to access actor state in addService, resulting in an async and probably throwing function to signal things like "already finished". Cleaner in a way (back-pressure, error-channel for funky states, ...) but not exactly a friendly API to call...

Any thoughts?

[czechboy0]

Hi @sliemeobn,

but not exactly a friendly API to call

can you say more about why it wouldn't be a friendly API to call?

[sliemeobn]

I feel like we'd be offloading a bunch of fringe details to the caller of addService who'd have to try await each time and somehow deal with errors that are at best mildly interesting.

Then again, async-ness probably would not be a big burden for users of ServiceGroup. Maybe an async function with a discardable result could be a compromise (similar to YieldResult)?

[FranzBusch]

I knew this moment would come to bite me and now it did. I regret making the ServiceGroup and actor. Now we can't change this but I think we can work around this. It should be possible to move all the state of the group into a nonisolated mutex and make all current methods nonisolated async. This should then unblock you to make your addService method nonisolated sync.

I agree with your general statement that addService shouldn't be async. I also think your approach with the stream and dynamically adding is the right one.

[sliemeobn]

I agree with your general statement that addService shouldn't be async.

@FranzBusch Now that I am a few LOC in, I am not so sure anymore.

Having an async func combined with an AsyncChannel makes for a pretty clean handling, without leaky buffers.
And addService is not exactly a high-throughput pipeline I would say 🤔.

    /// The internal state of the ``ServiceGroup``.
    private enum State {
        ...
        case running(
            gracefulShutdownStreamContinuation: AsyncStream<Void>.Continuation,
            addedServiceChannel: AsyncChannel<ServiceGroupConfiguration.ServiceConfiguration>
        )
        ....
    }

    public func addService(_ serviceConfiguration: ServiceGroupConfiguration.ServiceConfiguration) async {
        switch self.state {
        case var .initial(services: services):
            self.state = .initial(services: [])
            services.append(serviceConfiguration)
            self.state = .initial(services: services)

        case .running(_, let addedServiceChannel):
            await addedServiceChannel.send(serviceConfiguration)

        case .finished:
            return //TBD: I think doing nothing is fine, it is not unlike creating a service group and never calling run...
        }
    }

[sliemeobn]

I just opened a PR with the least intrusive method I could think of.
let me know what you think.

[FranzBusch]

Thanks @sliemeobn for the fix in #199