Readonly streams/partitions

[Jmgr]

This PR is a feature request and an API proposal to setting a readonly flag to streams and partitions.

Use-case

Our use-case involves the use of Liftbridge as part of the message processing pipeline in a future evolution of the Ably service. We currently implement elasticity of the service by allowing streams to be relocated - that is, the placement of a responsibility for a stream within an elastic cluster can move from time to time. We want to be able to implement a single, continuous, virtual stream, from multiple individual Liftbridge streams that each represent an epoch and are active at a given location for a period of time.

In order to be able to synchronise reliably from the ending of one epoch and the start of another, we would like to be able to put a stream into a read-only state, whereby subscribers can read from the stream, but no new messages can be published. Putting a stream into the read-only state needs to be idempotent, and a subscriber who reads to the end of a read-only stream is notified with an EOF-like error instead of blocking indefinitely.

This PR is a draft proposal for an API to set this state on a stream/partitions and also a possible implementation.

Further features

Other features we would like to have:

• having paused and readonly statuses to the PartitionMetadata returned by a FetchMetaData call ;
• expose the Raft commit index (or a timestamp) in FetchMetaData and in the activity stream, so that the client can ensure the validity of the data it is receiving ;
• add a timestamp to publication ACK's.

Current issues/limitations

While implementing this feature we have encountered some issues that we have not been able to fix yet, but we would welcome any advice on this feature request nevertheless and/or suggestions for fixes.

• no ErrEndOfReadonlyPartition error is sent when subscribing to a readonly partition that does not contain any message ;
• there is no unit test to check that subscribers that have subscribed before a partition is made readonly receive the ErrEndOfReadonlyPartition error, but local testing has shown that this feature is working ;
• in order to prevent existing subscribers that are already blocking (waiting for new data to arrive) to wait infinitely when a stream is set as read-only, some relatively extensive changes were needed (see the noBlocking channel). There might be a better solution to this.

Proposal

New RPCs to the API service:

// SetStreamReadonly sets a read-only flag to a partition. The latest
// message's offset is returned. Returns a NoSuchStream error code if the
// given stream or partition does not exist, and a PropertyAlreadySet error
// if this partition is already read-only.
rpc SetStreamReadonly(SetStreamReadonlyRequest) returns (SetStreamReadonlyResponse) {}

New messages:

// SetStreamReadonlyRequest is send to set a stream as read-only.
message SetStreamReadonlyRequest {
    string name               = 1; // Stream name
    repeated int32 partitions = 2; // Stream partitions
    bool readonly             = 3; // Should the readonly flag be set or removed
}

// SetStreamReadonlyResponse is sent by server after setting a stream's readonly
// flag.
message SetStreamReadonlyResponse {
    // Intentionally empty.
}

[tylertreat]

Interesting use case, thanks for the detailed proposal as always.

One thought that comes to mind is would it make more sense to have a generalized RPC for modifying stream configuration rather than specific RPCs, such as SetStreamReadonly? This would pave the way for changing various pieces of stream configuration dynamically after a stream is created. To start, this would just include a new "readonly" flag. This would also allow us to avoid having a bunch of config-specific RPCs as we introduce more settings in the future, such as authorization, re-partitioning, tiered storage, etc. I imagine as more features get introduced over time, this could become quite unwieldy from an API perspective.

Here is a sketch of what I'm thinking:

// SetStreamConfig dynamically updates a stream's configuration.
rpc SetStreamConfig(SetStreamConfigRequest) returns (SetStreamConfigResponse) {}

// SetStreamConfigRequest is sent to dynamically update a stream's configuration.
message SetStreamConfigRequest {
    string stream             = 1; // Name of stream to update
    repeated int32 partitions = 2; // Stream partitions to update, if applicable
    NullableBool readonly     = 3; // Enable/disable readonly
}

// SetStreamConfigResponse is sent by server after updating a stream's configuration.
message SetStreamConfigResponse {
    // Possibly include a mechanism for returning specific errors if it makes sense?
}

We would use NullableX types in order to distinguish between when a config is explicitly set or not (similar to how stream-level config works on CreateStream).

The only challenge I can foresee with this approach is around error handling, i.e. when there are multiple config values set on a request, how do we handle the case when some of them fail? This is where having individual RPCs is cleaner.

I have no idea if this approach is better or not, but thought it was worth considering.

[Jmgr]

One thought that comes to mind is would it make more sense to have a generalized RPC for modifying stream configuration rather than specific RPCs, such as SetStreamReadonly?

That would make sense, yes. "paused" could also be a flag.

The only challenge I can foresee with this approach is around error handling, i.e. when there are multiple config values set on a request, how do we handle the case when some of them fail? This is where having individual RPCs is cleaner.

Good point. Maybe with an array of results, one per partition? It would make the API less intuitive though.

[tylertreat]

Maybe with an array of results, one per partition? It would make the API less intuitive though.

This was my thought as well. Probably an array whose length equals the number of flags set on the request. It's a bit cumbersome, but I think a single API call is probably better? Plus, it means we can apply all of the changes to the Raft FSM at once.

[Jmgr]

Probably an array whose length equals the number of flags set on the request. It's a bit cumbersome, but I think a single API call is probably better? Plus, it means we can apply all of the changes to the Raft FSM at once.

Sounds good.

[Jmgr]

Have you made any progress on the generalized RPC for modifying stream configuration?

[tylertreat]

Have you made any progress on the generalized RPC for modifying stream configuration?

Planning to tackle this later this week.

[tylertreat]

Here are the WIP SetStreamConfig branches:

https://github.com/liftbridge-io/liftbridge-api/tree/set_stream_config https://github.com/liftbridge-io/liftbridge/tree/set_stream_config

If this reaches master I plan to keep it as a "beta" endpoint so that it can be changed without fear of breaking users until we're confident with the API. Feedback is welcome.

[tylertreat]

@Jmgr After playing around with the idea of a generalized RPC for stream config, I'm thinking we do not want to go that route, at least for now. It just feels too messy from an API perspective. Individual RPCs seems cleaner.

With that said, I'm good with bringing SetStreamReadonly in.

[Jmgr]

@Jmgr After playing around with the idea of a generalized RPC for stream config, I'm thinking we do not want to go that route, at least for now. It just feels too messy from an API perspective. Individual RPCs seems cleaner.

Alright, that makes sense.