Write an acknowledgement that member has processed earlier timestamps. Only the latest timestamp needs to be stored. Earlier timestamps can be overwritten. Acknowledgements of earlier timestamps should be ignored.

This takes the timestamp that is considered safe to prune from and walks it back to ensure that we have retained enough events to be able to start subscriptions from the counter checkpoint immediately preceding the timestamp.

Blocks until all earlier tasks have completed and then prevents further tasks from being run.

Count records currently stored by the sequencer. Used for pruning tests.

Delete all events that are ahead of the watermark of this sequencer. These events will not have been read and should be removed before returning the sequencer online. Should not be called alongside updating the watermark for this sequencer and only while the sequencer is offline.

Prevents member from sending and reading from the sequencer, and allows unread data for this member to be pruned. It however won't stop any sends addressed to this member.

Fetch a checkpoint with a counter value less than the provided counter.

Fetch the lower bound of events that can be read. Returns None if all events can be read.

Fetch the indexes of all sequencers that are currently online

Read the watermark for this sequencer and its online/offline status. Currently only used for testing.

Flag that we're going offline (likely due to a shutdown)

Mark the sequencer as online and return a timestamp for when this sequencer can start safely producing events.

Check whether we're closing. Susceptible to race conditions; unless you're using using this as a flag to the retry lib or you really know what you're doing, prefer performUnlessClosing and friends.

Check whether the member is enabled. Currently used when receiving a request for reading.

track running futures on shutdown

set to true to get detailed information about all futures that did not complete during shutdown. if set to false, we don't do anything.

Return the latest acknowledgements for all members

Lookup an existing member id for the given member. Will return a cached value if available. Return scala.None if no id exists.

Lookup member directly without caching.

Flag any sequencers that have a last updated watermark on or before the given cutoffTime as offline.

Performs the task given by f unless a shutdown has been initiated. The shutdown will only begin after f completes, but other tasks may execute concurrently with f, if started using this function, or one of the other variants (performUnlessClosingF and performUnlessClosingEitherT). The tasks are assumed to take less than closingTimeout to complete.

DO NOT CALL this.close as part of f, because it will result in a deadlock.

Performs the EitherT[Future] given by etf unless a shutdown has been initiated, in which case the provided error is returned instead. Both etf and the error are lazy; etf is only evaluated if there is no shutdown, the error only if we're shutting down. The shutdown will only begin after etf completes, but other tasks may execute concurrently with etf, if started using this function, or one of the other variants (performUnlessClosing and performUnlessClosingF). The tasks are assumed to take less than closingTimeout to complete.

DO NOT CALL this.close as part of etf, because it will result in a deadlock.

Performs the Future given by f unless a shutdown has been initiated. The future is lazy and not evaluated during shutdown. The shutdown will only begin after f completes, but other tasks may execute concurrently with f, if started using this function, or one of the other variants (performUnlessClosing and performUnlessClosingEitherT). The tasks are assumed to take less than closingTimeout to complete.

DO NOT CALL this.close as part of f, because it will result in a deadlock.

Prune as much data as safely possible from before the given timestamp. Return a human readable report on what has been removed.

Prune counter checkpoints for the given member before the given timestamp.

Prune events before the given timestamp

Prune payloads before the given timestamp

Read all events of which a member is a recipient from the provided timestamp but no greater than the earliest watermark.

Register the provided member. Should be idempotent if member is already registered and return the existing id.

Register a task that will run onClose after all "performUnlessShutdown" tasks have finished

Save a checkpoint that as of a certain timestamp the member has this counter value. Any future subscriptions can then use this as a starting point for serving their event stream rather than starting from 0.

Save a series of events to the store. Callers should determine batch size. No batching is done within the store. Callers MUST ensure that event-ids are unique as no errors will be returned if a duplicate is present (for the sequencer writer see sequencer.PartitionedTimestampGenerator for use with their instance index).

Save an updated lower bound of events that can be read. Must be equal or greater than any prior set lower bound.

Save the provided payloads to the store.

For DB implementations we suspect that this will be a hot spot for performance primarily due to size of the payload content values being inserted. To help with this we use storage.queryAndUpdateUnsafe to do these inserts using the full connection pool of the node instead of the single connection that is protected with an exclusive lock in the HA sequencer setup.

The downside of this optimization is that if a HA writer was to lose its lock, writes for payloads will continue regardless for a period until it shuts down. During this time another writer with the same instance index could come online. As we generate the payload id using a value generated for a partition based on the instance index this could worse case mean that two sequencer writers attempt to insert different payloads with the same id. If we use a simple idempotency method of just ignoring conflicts this could result in the active sequencer writing an event with a different payload resulting in a horrid corruption problem. This will admittedly be difficult to hit, but possible all the same.

The approach we use here is to generate an ephemeral instance discriminator for each writer instance.

• We insert the payloads using a simple insert (intentionally not ignoring conflicts unlike our other idempotent inserts). If this was successful we end here.
• If this insert hits a payload with the same id the query will blow up with a unique constraint violation exception. Now there's a couple of reasons this may have happened:
  1. Another instance is running and has inserted a conflicting payload (bad!) 2. There was some connection issue when running this query and our storage layer didn't see a successful result so it retried our query unaware that it did actually succeed so now we conflicting with our own insert. (a bit of a shame but not really a problem).
• We now query filtered on the payload ids we're attempting to insert to select out which payloads exist and their respective discriminators. For any payloads that exist we check that the discriminators indicate that we inserted this payload, if not a SavePayloadsError.ConflictingPayloadId error will be returned.
• Finally we filter to payloads that haven't yet been successfully inserted and go back to the first step attempting to reinsert just this subset.

Write the watermark that we promise not to write anything earlier than. Does not indicate that there is an event written by this sequencer for this timestamp as there may be no activity at the sequencer, but updating the timestamp allows the sequencer to indicate that it's still alive. Return an error if we find our sequencer is offline.

Build a status object representing the current state of the sequencer.

Validate that the commit mode of a session is inline with the configured expected commit mode. Return a human readable message about the mismatch in commit modes if not.