Re-using connection for long lived commands like subscriptions

[StefanPrintezis]

In the pubsub docs example when subscribing you're supposed to use the .withConnection.

try await valkeyClient.withConnection { connection in
    try await connection.subscribe(to: ["channel1", "channel2"]) { subscription in
        for try await item in subscription {
            // a subscription item includes the channel the message was published on
            // as well as the message
            print(item.channel)
            print(item.message)
        }
    }
}

What would your advice be when you want to subscribe, but don't need to claim the connection in it's closure? Effectively re-using a connection for long lived subscriptions?

Right now i've implemented my own "connection manager" & keep withConnections closures open during a service run() so my thousands of subscriptions can run on just several connections, but it feels like i'm circumventing the design somehow or i'm missing something here, so some direction on this part would be greatly appreciated!

Also before my implementation ("connection manager") I depleted my connections pretty quickly in my tests but did not have an easy way of telling the connection pool could no longer provide an available connection. Should I see a log statement if the command times out? I probably don't see it because other timeouts trigger sooner I think

[StefanPrintezis]

#162

connections to get used across multiple tasks

[adam-fowler]

You can use a task group to run multiple subscriptions on the same connection

eg

try await valkeyClient.withConnection { connection in
    try await withThrowingTaskGroup { group in
        group.addTask {
            try await connection.subscribe(to: ["channel1", "channel2"]) { subscription in
                ...
            }
        }
        group.addTask {
            try await connection.subscribe(to: "channel3") { subscription in
                ...
            }
        }
}

What you could do is have a subscription manager which runs a top level process and accepts subscription requests

struct SubscriptionManager: Service {
    let (stream, cont) = AsyncStream.makeStream(of: (String, @Sendable (ValkeySubscription) async throws -> Void).self)
    func run() async throws {
        try await valkeyClient.withConnection { connection in
            try await withDiscardingTaskGroup { group in
                for try await sub in self.stream {
                    group.addTask {
                        try await connection.subscribe(to: sub.0) { subscription in
                            try await sub.1(subscription)
                        }
                    }
                }
            }
        }
    }

Or something like that. I haven't tested this code at all

[adam-fowler]

We also have a client version of subscribe PR currently being reviewed which will deal with disconnects.

EDIT: That won't help you with using one connection for multiple subscriptions

[StefanPrintezis]

I think I was looking for a distinction between a connection being either available or leased. The connection shouldn't be idle if there is an active subscription, but it should be available for other commands. Keeping the connection leased to my manager didn't make sense to me in the way that i'd need at least 2 connections to write application logic while e.g. a maximum of 1 would be valid configuration.

Long-lived commands create room for a leased available connection if that makes sense?

[Joannis]

@adam-fowler in the current setup, there's no practical way to have one Redis connection reused between multiple Services if one of the services needs to subscribe on that connection. For the duration of the subscription, that connection will be leased. Whilst it is leased, other services won't be able to lease that same connection. You can technically work around this with a task group, but architecturally that'll be a nightmare. You don't want your EventSubscriptionService to handle SET (or other) calls into Redis on behalf of another Service. But you would want the connection to be reusable while it's handling subscription data - because RESP supports that.

[adam-fowler]

The problem with making a connection running a subscription available to run commands in a generic manner, is a user could lease that connection and then run a blocking command like BRPOP. This would block the subscription

EDIT: I was wrong it doesn't block the subscription.

[fabianfett]

@StefanPrintezis I totally get your point, that the current API forces users of this library to manage connections themselves, as they would otherwise run out of connections, if they do a lot of subscribing.

I think the missing feature the client needs here is a hidden subscription connection, so that ValkeyClient can offer subscribe calls and all those subscriptions get put on the same connection. If this connection is closed, the client can auto recreate all the subscriptions on it.

I think I was looking for a distinction between a connection being either available or leased. The connection shouldn't be idle if there is an active subscription, but it should be available for other commands.

While I understand this technical perspective it brings us into an area where we suddenly have three connection types: idle, leased, idleButSubscribed. This makes a connection pool, a harder to implement, b harder to reason about, c less reusable.

Do you think you could live with the single hidden subscription connection in the client?

[StefanPrintezis]

How about MONITOR, PSUBSCRIBE, or any blocking list operation BLPOP. If all of these lock on to a connection until they've matched on a response then we'll be running out of leasable connections pretty quickly.

I wouldn't go for a leased+locked state, but assume a connection can be waiting for reads. Maybe idle active leased

Having just 1 connection would not fully utilize the hardware and i'm not fully aware how far that would be an issue when using a cluster.

I'm afraid for the most performant option I would need another connection state, which is what i'm trying to recreate in my "connection manager" implementation, but it feels like that should be part of the connection itself.

Right now when i have more than 0 subscriptions i lease a connection and release the connection when the subscription count reaches 0. My next step would be to have a pool of connections leased as my subscriptions increase so I would be balancing them out.

I'm even thinking of creating 2 clients where the first would just be entirely consumed by my "SubscriptionConnectionManager" and the other for "normal" usage so there's no conflict there (which partly aligns with the hidden connection suggestion you made). At this point I'd consider instead of wrapping around the connection pool in this way, which is not per se very maintainable, to just fully manage connection state which brings me back to the library.

[adam-fowler]

I've updated the ValkeyClient.subscribe PR #131 to only use one connection for all subscriptions, that includes PSUBSCRIBE.