Feature request: PinnedByCorrelation support in gRPC clients

[innocentiv]

According to documentation KurrentDB supports PinnedByCorrelation Consumer Strategy for Persistent Subscription: https://docs.kurrent.io/server/v25.0/features/persistent-subscriptions.html#consumer-strategies

This Consumer Strategy enable us to drastically reduce the amount of concurrency issues that may arise when processing sagas across multiple streams.

But I have noticed that this consumer strategy is not supported in the node client and only partially supported in the web UI and in the HTTP client (not supported in the $all subscription but supported in the stream subscription as $all subscription is not supported in HTTP client)

The gRPC protobuf schema on the server is inconsistent between creation and update and contains two separate way to update the consumer strategy:

ConsumerStrategy named_consumer_strategy = 13 [deprecated = true]
string consumer_strategy = 16;

which are defined in the schema type as

    enum ConsumerStrategy {
        DispatchToSingle = 0;
        RoundRobin = 1;
        Pinned = 2;
    }

and in the server codebase as

public static class SystemConsumerStrategies {
    public const string DispatchToSingle = "DispatchToSingle";
    public const string RoundRobin = "RoundRobin";
    public const string Pinned = "Pinned";
    public const string PinnedByCorrelation = "PinnedByCorrelation";
}

respectively.

When using consumer_strategy in the node client instead of named_consumer_strategy (deprecated), I can actually set the proper 'PinnedByCorrelation' strategy:

// node_modules/@kurrent/kurrentdb-client/dist/persistentSubscription/utils/settingsToGRPC.js

case constants_1.ROUND_ROBIN: {
    reqSettings.setNamedConsumerStrategy(persistent_pb_1.CreateReq.ConsumerStrategy.ROUNDROBIN);
    break;
}
// ADD THIS NEW CASE BY USING CONSUMER STRATEGY INSTEAD OF NAMED CONSUMER STRATEGY
case constants_1.PINNED_BY_CORRELATION: {
    reqSettings.setConsumerStrategy("PinnedByCorrelation");
    break;
}
default: {
    console.warn(`Unknown consumerStrategyName ${settings.consumerStrategyName}.`);
    break;
}

This is affecting both the subscription to $all and the subscription to stream. You can find here a fork with the modification I have made to the client to make it behave correctly for my usecase: master...innocentiv:KurrentDB-Client-NodeJS:master

This also apply to other clients as well, but I have opened this in Node repo as this is what we use.

PinnedByCorrelation is the most useful strategy for us, and looks like both the server schema and the clients needs some work to actually support it correctly. Is this going to be fixed in the future?

[stktung]

@alexeyzimarev

[w1am]

@innocentiv I can bring this feature to the clients, as long as we agree that you won’t be able to update the consumer strategy to PinnedByCorrelation, since the server doesn’t support it, and we make sure this is well documented. I will asked them to fix this issue with updates in the db.

[innocentiv]

@w1am what do you mean that the server doesn't support it? You can set it in the server after creation if you want to. And with the changes in the fork I shared above it works also with the node client. Do you mean to say that even if the strategy is set for the persistent prescription it is not actually working?

I see the implementation here:

public class PinnedByCorrelationPersistentSubscriptionConsumerStrategy : PinnedPersistentSubscriptionConsumerStrategy {

	public PinnedByCorrelationPersistentSubscriptionConsumerStrategy(IHasher<string> streamHasher) : base(streamHasher) {
	}

	public override string Name {
		get { return SystemConsumerStrategies.PinnedByCorrelation; }
	}


	protected override string GetAssignmentSourceId(ResolvedEvent ev) {
		var eventRecord = ev.Event ?? ev.Link;

		string correlation = CorrelationFromJsonBytes(eventRecord.Metadata);

		if (correlation == null) {
			return GetSourceStreamId(ev);
		}

		return correlation;
	}

that is actually getting the Hash from the correlationId

	private string CorrelationFromJsonBytes(ReadOnlyMemory<byte> toConvert) {
		using (var reader = new JsonTextReader(new StreamReader(new MemoryStream(toConvert.ToArray())))) {
			if (!reader.Read()) {
				return null;
			}

			while (true) {
				if (!reader.Read()) {
					return null;
				}

				if (reader.TokenType == JsonToken.EndObject) {
					break;
				}

				if (reader.TokenType == JsonToken.PropertyName) {
					if ((string)reader.Value == CorrelationIdPropertyContext.CorrelationIdProperty) {
						reader.Read();

						if (reader.TokenType == JsonToken.String) {
							return (string)reader.Value;
						}
					}
				}
			}
		}

		return null;
	}

why do you say this is not working?

[w1am]

It does support it, but only during creation, not on updates. What I’m trying to say is that you can create a persistent subscription with the PinnedByCorrelation strategy, but you won’t be able to update it afterward.

[innocentiv]

Ok, I understand your point better now and we agree. I think this is just a problem with the schema though. There is no intrinsic reason why "PinnedByCorrelation" is not supported in UpdateReq, and just a mismatch between the schema and the underlying handler. HTTP client support "PinnedByCorrelation" on update already. The problem to solve is to fix the behaviour while maintaining backwards compatibility between clients and servers versions.

[w1am]

I will need to check, but there are features that are not yet supported in the gRPC operations for projections as well.
In the client’s case, I think falling back to using HTTP will solve the issue.