Improve rewind interface for streaming clients

[Theodus]

We should provide an abstraction that simplifies the logic necessary for streaming clients to handle rewinds (docs). Rather than providing a set of ranges, rewinds/invalidation should be based on equality comparisons against watermarks.

[leoyvens]

Concretely app_metadata for reorg messages should include the end watermark of the previously sent segment.

[leoyvens]

I'm now noticing we actually already have all the necessary information in the app_metadata, but it's non-trivial to interpret, particularly when generalizing to multi-chain.

Another problem with the current protocol is the end of the range being pre-estabilished, which prevents the server from dynamically emitting checkpoints (after a time or bytes budget is exceeded).

I have some ideas for a V2 of our streaming protocol.

Cursor Definition
Let’s define a cursor as an opaque value that a client can send back in the amp-resume header to resume from exactly that point. A cursor may internally contain multi-chain positions (block number, hash...), but clients don’t have to understand it to use it.

The protocol then consists of three message kinds. All of them carry a cursor, but the semantics differ:

1. Data messages. These carry a next_checkpoint in app_metadata. This is a cursor for a future point at which the server knows it will emit a checkpoint. The server might still emit early checkpoints before the known one, as an optimization to allow the client to flush data.
2. Checkpoints. Represents completeness up to a given cursor present in app_metadata. This is what the client would store to resume from. This message might also carry finality information when we start tracking that.
3. Rewinds. These have a back_to_checkpoint field, which is the checkpoint cursor to rewind to. Used to express reorgs, fork changes or retraction of optimistic updates.

This protocol should be intuitive to interpret even by systems that don't know what a blockchain is, as cursors can be treated as opaque values. If the downstream system does want deep blockchain semantics, you can inspect the cursor field to extract block numbers, hashes and other information.

Handling Rewinds

When receiving a rewind, the client can choose its strategy:

1. Transaction based: Store previous cursor updates and associated them with your 'transaction ids'. If the rewind cursor field is equal to the cursor of one of your previous transactions, undo all later transactions and continue.
2. Block-range based: If you are storing the row watermarks, you can inspect the cursor payload, derive the corresponding watermark, and delete all rows with a greater watermark. See 'Row watermarks' in the next section.

Row watermarks

Each row in a dataset carries a watermark, currently represented by the _block_num column. This is exact for each row, so systems wanting to track time information for individual rows can track this value.

[fubhy]

Block-range based: If you are storing the row watermarks, you can inspect the cursor payload, derive the corresponding watermark, and delete all rows with a greater watermark. See 'Row watermarks' in the next section.

How would you define "with a greater watermark"? Watermarks are not sortable. This is basically why I introduced the monotonically increasing transaction ids. Both as transaction identifiers obviously but also to make "range" semantics easier for a consumer to invalidate data downstream.

How would this work in a multichain world where you don't have a single block num?

[leoyvens]

@fubhy I'm proposing a terminology change here. What we currently call 'watermark' is now called 'cursor'. The term 'watermark' is now used for the _block_num height associated to the rows, and cursors also have watermarks. We can have a cursor.watermark field.

Watermarks are sortable, and figuring out how to maintain that property when we generalize beyond a single _block_num will be a crux of multichain support. This proposal is trying to prepare the server protocol to accommodate multichain, but solving multichain is out of scope.

[leoyvens]

Each message could also carry a msg_count metadata field, which is the number of messages previously emitted on that same connection. A small thing to help traceability and help clients track tx counters on their side.