From 0e812f6741a85d85aa1b05d42a3f7111bc300ef0 Mon Sep 17 00:00:00 2001 From: alnkesq <141324281+alnkesq@users.noreply.github.com> Date: Tue, 24 Mar 2026 19:19:27 +0100 Subject: [PATCH 1/3] LastDefinitelyProcessedCursor for JetStream and ATProto clients --- src/FishyFlip/ATJetStream.cs | 58 ++++++++----- src/FishyFlip/ATWebSocketProtocol.cs | 86 ++++++++++++-------- src/FishyFlip/OutOfOrderCompletionTracker.cs | 71 ++++++++++++++++ 3 files changed, 162 insertions(+), 53 deletions(-) create mode 100644 src/FishyFlip/OutOfOrderCompletionTracker.cs diff --git a/src/FishyFlip/ATJetStream.cs b/src/FishyFlip/ATJetStream.cs index adaa3042..e2bd561a 100644 --- a/src/FishyFlip/ATJetStream.cs +++ b/src/FishyFlip/ATJetStream.cs @@ -21,6 +21,7 @@ public sealed class ATJetStream : IDisposable private readonly JsonSerializerOptions jsonSerializerOptions; private readonly SourceGenerationContext sourceGenerationContext; + private readonly OutOfOrderCompletionTracker outOfOrderCompletionTracker = new(); private ClientWebSocket client; private bool disposedValue; private ILogger? logger; @@ -60,6 +61,11 @@ public ATJetStream(ATJetStreamOptions options) /// public event EventHandler? OnRecordReceived; + /// + /// Gets the last definitely processed firehose cursor. + /// + public long? LastDefinitelyProcessedCursor => this.outOfOrderCompletionTracker.LastDefinitelyProcessedSeq; + /// void IDisposable.Dispose() { @@ -286,7 +292,8 @@ private async Task ReceiveMessages(ClientWebSocket webSocket, CancellationToken var message = Encoding.UTF8.GetString(messageBytes); #endif this.OnRawMessageReceived?.Invoke(this, new JetStreamRawMessageEventArgs(message)); - this.options.TaskFactory.StartNew(() => this.HandleMessage(message)) + var eventId = this.outOfOrderCompletionTracker.OnEventGenerated(); + this.options.TaskFactory.StartNew(() => this.HandleMessage(message, eventId)) .FireAndForgetSafeAsync(this.logger); } catch (OperationCanceledException) @@ -302,35 +309,44 @@ private async Task ReceiveMessages(ClientWebSocket webSocket, CancellationToken this.OnConnectionUpdated?.Invoke(this, new SubscriptionConnectionStatusEventArgs(webSocket.State)); } - private void HandleMessage(string json) + private void HandleMessage(string json, long eventId) { - if (string.IsNullOrEmpty(json)) - { - this.logger?.LogDebug("WSS: Empty message received."); - return; - } - + long? seq = null; try { - var atWebSocketRecord = JsonSerializer.Deserialize(json, this.sourceGenerationContext.ATWebSocketRecord); - if (atWebSocketRecord is null) + if (string.IsNullOrEmpty(json)) { - this.logger?.LogError("WSS: Failed to deserialize ATWebSocketRecord."); - this.logger?.LogError(json); + this.logger?.LogDebug("WSS: Empty message received."); return; } - this.OnRecordReceived?.Invoke(this, new JetStreamATWebSocketRecordEventArgs(atWebSocketRecord, json)); - } - catch (JsonException ex) - { - this.logger?.LogError(ex, "WSS: Failed to deserialize ATWebSocketRecord."); - this.logger?.LogError(json); + try + { + var atWebSocketRecord = JsonSerializer.Deserialize(json, this.sourceGenerationContext.ATWebSocketRecord); + if (atWebSocketRecord is null) + { + this.logger?.LogError("WSS: Failed to deserialize ATWebSocketRecord."); + this.logger?.LogError(json); + return; + } + + seq = atWebSocketRecord.TimeUs; + this.OnRecordReceived?.Invoke(this, new JetStreamATWebSocketRecordEventArgs(atWebSocketRecord, json)); + } + catch (JsonException ex) + { + this.logger?.LogError(ex, "WSS: Failed to deserialize ATWebSocketRecord."); + this.logger?.LogError(json); + } + catch (Exception ex) + { + this.logger?.LogError(ex, "WSS: An unknown error occurred."); + this.logger?.LogError(json); + } } - catch (Exception ex) + finally { - this.logger?.LogError(ex, "WSS: An unknown error occurred."); - this.logger?.LogError(json); + this.outOfOrderCompletionTracker.OnEventProcessed(eventId, seq); } } diff --git a/src/FishyFlip/ATWebSocketProtocol.cs b/src/FishyFlip/ATWebSocketProtocol.cs index be8ecdab..cf88192e 100644 --- a/src/FishyFlip/ATWebSocketProtocol.cs +++ b/src/FishyFlip/ATWebSocketProtocol.cs @@ -13,6 +13,7 @@ namespace FishyFlip; public sealed class ATWebSocketProtocol : IDisposable { private readonly TaskFactory taskFactory; + private readonly OutOfOrderCompletionTracker outOfOrderCompletionTracker = new(); private bool subscribedLabels; private WebSocketWrapper webSocketWrapper; private IReadOnlyList customConverters; @@ -73,6 +74,11 @@ public ATWebSocketProtocol(ATWebSocketProtocolOptions options) /// public bool IsConnected => this.webSocketWrapper.IsConnected; + /// + /// Gets the last definitely processed firehose cursor. + /// + public long? LastDefinitelyProcessedCursor => this.outOfOrderCompletionTracker.LastDefinitelyProcessedSeq; + /// /// Connect to the BlueSky instance via a WebSocket connection. /// @@ -205,48 +211,56 @@ private void Dispose(bool disposing) } } - private void HandleMessage(ReadOnlySpan span) + private void HandleMessage(ReadOnlySpan span, long eventId) { - byte[] byteArray = span.ToArray(); - if (byteArray.Length == 0) - { - this.logger?.LogDebug("WSS: ATError reading message. Empty byte array."); - return; - } - - CBORObject[]? objects = null; + long? seq = null; try { - objects = CBORObject.DecodeSequenceFromBytes(byteArray, new CBOREncodeOptions("useIndefLengthStrings=true;float64=true;allowduplicatekeys=true;allowEmpty=true")); - } - catch (Exception e) - { - this.logger?.LogError(e, "WSS: ATError reading message."); - } + byte[] byteArray = span.ToArray(); + if (byteArray.Length == 0) + { + this.logger?.LogDebug("WSS: ATError reading message. Empty byte array."); + return; + } - if (objects is null) - { - return; - } + CBORObject[]? objects = null; + try + { + objects = CBORObject.DecodeSequenceFromBytes(byteArray, new CBOREncodeOptions("useIndefLengthStrings=true;float64=true;allowduplicatekeys=true;allowEmpty=true")); + } + catch (Exception e) + { + this.logger?.LogError(e, "WSS: ATError reading message."); + } - if (objects.Length != 2) - { - return; - } + if (objects is null) + { + return; + } - var frameHeader = new FrameHeader(objects[0]); + if (objects.Length != 2) + { + return; + } - if (this.subscribedLabels) - { - this.HandleLabelMessage(frameHeader, objects[1]); + var frameHeader = new FrameHeader(objects[0]); + + if (this.subscribedLabels) + { + seq = this.HandleLabelMessage(frameHeader, objects[1]); + } + else + { + seq = this.HandleRepoMessage(frameHeader, objects[1]); + } } - else + finally { - this.HandleRepoMessage(frameHeader, objects[1]); + this.outOfOrderCompletionTracker.OnEventProcessed(eventId, seq); } } - private void HandleLabelMessage(FrameHeader header, CBORObject obj) + private long? HandleLabelMessage(FrameHeader header, CBORObject obj) { var message = new SubscribeLabelMessage(); message.Header = header; @@ -281,9 +295,10 @@ private void HandleLabelMessage(FrameHeader header, CBORObject obj) } this.OnSubscribedLabelMessage?.Invoke(this, new SubscribedLabelEventArgs(message)); + return message.Labels?.Seq; } - private void HandleRepoMessage(FrameHeader header, CBORObject obj) + private long? HandleRepoMessage(FrameHeader header, CBORObject obj) { var message = new SubscribeRepoMessage(); message.Header = header; @@ -377,15 +392,22 @@ private void HandleRepoMessage(FrameHeader header, CBORObject obj) } this.OnSubscribedRepoMessage?.Invoke(this, new SubscribedRepoEventArgs(message)); + + return message.Commit?.Seq; } private Task OnInternalMessageReceived(ReadOnlySequence message) { this.OnMessageReceived?.Invoke(this, message); + var eventId = this.outOfOrderCompletionTracker.OnEventGenerated(); if (this.OnRecordReceived is not null || this.OnSubscribedRepoMessage is not null || this.OnSubscribedLabelMessage is not null) { var newMessage = message.ToArray(); - this.taskFactory.StartNew(() => this.HandleMessage(newMessage)).FireAndForgetSafeAsync(this.logger); + this.taskFactory.StartNew(() => this.HandleMessage(newMessage, eventId)).FireAndForgetSafeAsync(this.logger); + } + else + { + this.outOfOrderCompletionTracker.OnEventProcessed(eventId, null); } return Task.CompletedTask; diff --git a/src/FishyFlip/OutOfOrderCompletionTracker.cs b/src/FishyFlip/OutOfOrderCompletionTracker.cs new file mode 100644 index 00000000..48fbe47d --- /dev/null +++ b/src/FishyFlip/OutOfOrderCompletionTracker.cs @@ -0,0 +1,71 @@ +// +// Copyright (c) Drastic Actions. All rights reserved. +// +using System.Runtime.CompilerServices; + +namespace FishyFlip +{ + /// + /// Keeps track of the most recent event ID such that all previous events have definitely been processed. + /// + internal class OutOfOrderCompletionTracker + { + private readonly Dictionary eventIdToSeq = new(); + private long lastDefinitelyProcessedEventId; + private long? lastDefinitelyProcessedSeq; + private long lastGeneratedEventId; + + /// + /// Gets the last firehose cursor such that this and all previous events have been already processed on the threadpool. + /// + public long? LastDefinitelyProcessedSeq + { + get + { + lock (this) + { + return this.lastDefinitelyProcessedSeq; + } + } + } + + /// + /// To be called when an event is generated but not processed yet. Each call must eventually result in a . + /// + /// A sequentially increasing event ID that must be eventually passed to . + [MethodImpl(MethodImplOptions.Synchronized)] + public long OnEventGenerated() + { + return ++this.lastGeneratedEventId; + } + + /// + /// To be called when an event is fully processed (successfully or unsuccessfully). Each call must come after a corresponding . + /// + /// The event ID. + /// The firehose cursor of the current entry, if available. + [MethodImpl(MethodImplOptions.Synchronized)] + public void OnEventProcessed(long eventId, long? seq) + { + if (eventId <= this.lastDefinitelyProcessedEventId) + { + throw new InvalidOperationException("OnEventProcessed was called with an eventId that was already supposedly processed."); + } + + this.eventIdToSeq.Add(eventId, seq); + this.UpdateLastDefinitelyProcessedEvent(); + } + + private void UpdateLastDefinitelyProcessedEvent() + { + while (this.eventIdToSeq.TryGetValue(this.lastDefinitelyProcessedEventId + 1, out var seq)) + { + this.lastDefinitelyProcessedEventId++; + this.lastDefinitelyProcessedSeq = seq; + this.eventIdToSeq.Remove(this.lastDefinitelyProcessedEventId); + } + } + } +} + + From 5a9f0572fac702eec8e3416d92ef2d45ec3ce606 Mon Sep 17 00:00:00 2001 From: alnkesq <141324281+alnkesq@users.noreply.github.com> Date: Tue, 24 Mar 2026 20:48:41 +0100 Subject: [PATCH 2/3] Don't overwrite with lastDefinitelyProcessedSeq with null if seq is not available --- src/FishyFlip/OutOfOrderCompletionTracker.cs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/FishyFlip/OutOfOrderCompletionTracker.cs b/src/FishyFlip/OutOfOrderCompletionTracker.cs index 48fbe47d..f5d5a6e5 100644 --- a/src/FishyFlip/OutOfOrderCompletionTracker.cs +++ b/src/FishyFlip/OutOfOrderCompletionTracker.cs @@ -61,7 +61,11 @@ private void UpdateLastDefinitelyProcessedEvent() while (this.eventIdToSeq.TryGetValue(this.lastDefinitelyProcessedEventId + 1, out var seq)) { this.lastDefinitelyProcessedEventId++; - this.lastDefinitelyProcessedSeq = seq; + if (seq != null) + { + this.lastDefinitelyProcessedSeq = seq; + } + this.eventIdToSeq.Remove(this.lastDefinitelyProcessedEventId); } } From a585981f250cfe37446d78d1da15db4d5ece811e Mon Sep 17 00:00:00 2001 From: alnkesq <141324281+alnkesq@users.noreply.github.com> Date: Tue, 24 Mar 2026 21:12:49 +0100 Subject: [PATCH 3/3] OnEventGenerated doesn't need locks, lastGeneratedEventId is only ever used there --- src/FishyFlip/OutOfOrderCompletionTracker.cs | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/FishyFlip/OutOfOrderCompletionTracker.cs b/src/FishyFlip/OutOfOrderCompletionTracker.cs index f5d5a6e5..731edf5d 100644 --- a/src/FishyFlip/OutOfOrderCompletionTracker.cs +++ b/src/FishyFlip/OutOfOrderCompletionTracker.cs @@ -33,10 +33,9 @@ public long? LastDefinitelyProcessedSeq /// To be called when an event is generated but not processed yet. Each call must eventually result in a . /// /// A sequentially increasing event ID that must be eventually passed to . - [MethodImpl(MethodImplOptions.Synchronized)] public long OnEventGenerated() { - return ++this.lastGeneratedEventId; + return Interlocked.Increment(ref this.lastGeneratedEventId); } /// @@ -44,16 +43,18 @@ public long OnEventGenerated() /// /// The event ID. /// The firehose cursor of the current entry, if available. - [MethodImpl(MethodImplOptions.Synchronized)] public void OnEventProcessed(long eventId, long? seq) { - if (eventId <= this.lastDefinitelyProcessedEventId) + lock (this) { - throw new InvalidOperationException("OnEventProcessed was called with an eventId that was already supposedly processed."); - } + if (eventId <= this.lastDefinitelyProcessedEventId) + { + throw new InvalidOperationException("OnEventProcessed was called with an eventId that was already supposedly processed."); + } - this.eventIdToSeq.Add(eventId, seq); - this.UpdateLastDefinitelyProcessedEvent(); + this.eventIdToSeq.Add(eventId, seq); + this.UpdateLastDefinitelyProcessedEvent(); + } } private void UpdateLastDefinitelyProcessedEvent()