feat(broker): G21-Followup C+D — WAL-Trim + Read-Fallback

Schliesst die dritte Closing-Bedingung aus #11 und ihren impliziten
Read-Path-Pre-Requirement. Operatoren koennen jetzt
TrimAfterFlush=true sicher aktivieren, weil getrimmte Offsets ueber
DisaggregatedSegmentReader aus dem Object-Store nachgefuettert werden.

Geaendert (Followup D — Read-Fallback):
- DataApiHandler ctor +DisaggregatedSegmentReader? disaggregatedReader.
  HandleFetchAsync konsultiert vor dem LogManager.ReadBatches*-Call:
  wenn _disaggregatedReader != null && topicMetadata.IsDisaggregated &&
  reader.TryReadAsync.HitManifest => Bytes aus Manifest+Remote zurueck,
  Skip der lokalen Read-Pfade. messageCount via batch-header-walk an
  Position 57. Sonst Fall-through zur existierenden Logik — Null-Default
  haelt Pre-G21-Verhalten 1:1.
- SurgewaveRuntimeOptions +DisaggregatedReader Property.
- SurgewaveRuntimeBuilder +WithDisaggregatedReader builder method.
- SurgewaveRuntime passt _options.DisaggregatedReader an DataApiHandler durch.

Geaendert (Followup C — Trim):
- WalSealedSegment +TrimAsync optional delegate (Func<CT, Task>).
- WalFlusherOptions +TrimAfterFlush flag (default false fuer Safety —
  Aktivieren ohne Read-Fallback wuerde getrimmte Offsets unreachable
  machen; Doc-Comment weist auf WithDisaggregatedReader hin).
- WalFlusher.FlushOneAsync ruft TrimAsync NACH erfolgreichem
  AppendObjectAsync. Trim-Exception wird geloggt + geswallowed
  (Manifest = Source-of-Truth, Stray-Local-File ist kostenlos bis
  zur naechsten Retention-Sweep).
- PartitionLogWalSegmentSource setzt TrimAsync auf File.Delete der
  3 Segment-Files (.log, .index, .timeindex; idempotent ueber
  DeleteIfExists-Helper).

Tests (4 neue, 57 gesamt, alle gruen):
- TrimAfterFlush=true + delegate set -> trimmed=true
- TrimAfterFlush=false + delegate set -> trimmed=false (knob respected)
- Failed upload -> TrimAsync NICHT aufgerufen (atomic-or-nothing)
- Trim-Exception nach erfolgreicher Manifest-Commit -> swallowed,
  Manifest bleibt 1 entry

Verifikation: Solution-Build clean, Disaggregated 57/57 + Protocol.Native
153/153 + Broker baut. Followups A+B+C+D = alle 3 Closing-Bedingungen
aus #11 durch — Issue kann jetzt geclosed werden.

Author: thomas-stegemann

src/Kuestenlogik.Surgewave.Broker/Handlers/DataApiHandler.cs

@@ -11,6 +11,7 @@
 using Kuestenlogik.Surgewave.Core.Util;
 using Kuestenlogik.Surgewave.Protocol.Kafka;
 using Kuestenlogik.Surgewave.Protocol.Kafka.Requests;
+using Kuestenlogik.Surgewave.Storage.Disaggregated.Read;
 using Kuestenlogik.Surgewave.Storage.Disaggregated.Routing;
 using Microsoft.Extensions.Logging;
 
@@ -36,6 +37,7 @@ public sealed class DataApiHandler : IKafkaRequestHandler
     private readonly IRecordTransformPipeline? _recordTransform;
     private readonly ColdStartWorkloadProfiler? _coldStartProfiler;
     private readonly IPartitionAppender _partitionAppender;
+    private readonly DisaggregatedSegmentReader? _disaggregatedReader;
     private readonly ILogger<DataApiHandler> _logger;
 
     public IEnumerable<ApiKey> SupportedApiKeys =>
@@ -61,7 +63,8 @@ public DataApiHandler(
         SurgewaveBrokerObservability? observability = null,
         IRecordTransformPipeline? recordTransform = null,
         ColdStartWorkloadProfiler? coldStartProfiler = null,
-        IPartitionAppender? partitionAppender = null)
+        IPartitionAppender? partitionAppender = null,
+        DisaggregatedSegmentReader? disaggregatedReader = null)
     {
         _config = config;
         _logManager = logManager;
@@ -82,7 +85,8 @@ public DataApiHandler(
         // enable disaggregated storage pass a RoutingPartitionAppender via
         // SurgewaveRuntimeBuilder.WithPartitionAppender(...).
         _partitionAppender = partitionAppender
-            ?? new DelegatingPartitionAppender((tp, batch, _, ct) => _logManager.AppendBatchAsync(tp, batch, ct));
+            ?? new DelegatingPartitionAppender((tp, batch, _, ct) => _logManager.AppendBatchAsync(tp, batch, ct).AsTask());
+        _disaggregatedReader = disaggregatedReader;
     }
 
     public async Task<KafkaResponse> HandleAsync(KafkaRequest request, RequestContext context, CancellationToken cancellationToken)
@@ -270,9 +274,9 @@ private async Task<ProduceResponse> HandleProduceAsync(ProduceRequest request, C
                     // StatelessAgent. The record count is parsed from the batch
                     // header (Kafka RecordBatch v2, offset 57); stateless mode
                     // needs it for offset assignment.
-                    var recordCount = RecordHeaderParser.ParseBatchHeader(recordsToAppend.Span).RecordCount;
+                    var produceRecordCount = RecordHeaderParser.ParseBatchHeader(recordsToAppend.Span).RecordCount;
                     var baseOffset = await _partitionAppender.AppendBatchAsync(
-                        topicPartition, recordsToAppend, recordCount, cancellationToken);
+                        topicPartition, recordsToAppend, produceRecordCount, cancellationToken);
 
                     // Register hash after successful write (deduplication)
                     _deduplicationManager?.Register(topicPartition, recordsToAppend.Span, baseOffset);
@@ -483,6 +487,52 @@ private async Task<FetchResponse> HandleFetchAsync(FetchRequest request, Connect
                     byte[] recordSet;
                     int messageCount;
 
+                    // Disaggregated read fallback: when the topic uses
+                    // disaggregated storage and the requested offset has
+                    // already been flushed to the object store (i.e. the
+                    // local WAL no longer holds it), serve from the
+                    // manifest. The reader returns HitManifest=false for
+                    // offsets past the manifest tail — those still live in
+                    // the local WAL and the normal read path below picks
+                    // them up. Skip when no reader is wired (default) or
+                    // when the topic isn't disaggregated.
+                    var fetchTopicMetadata = _logManager.GetTopicMetadata(topic);
+                    if (_disaggregatedReader is not null && fetchTopicMetadata?.IsDisaggregated == true)
+                    {
+                        var disagRead = await _disaggregatedReader.TryReadAsync(
+                            topicPartition,
+                            partitionData.FetchOffset,
+                            partitionData.MaxBytes,
+                            cancellationToken).ConfigureAwait(false);
+                        if (disagRead.HitManifest)
+                        {
+                            recordSet = disagRead.LogBytes.ToArray();
+                            messageCount = 0;
+                            // Same record-count tallying pattern as the
+                            // contiguous fast path: walk the concatenated
+                            // batches and read the count field at offset 57.
+                            var span = disagRead.LogBytes.Span;
+                            var cursor = 0;
+                            while (cursor + 61 <= span.Length)
+                            {
+                                var batchLen = System.Buffers.Binary.BinaryPrimitives.ReadInt32BigEndian(span.Slice(cursor + 8, 4));
+                                var batchTotal = 12 + batchLen; // baseOffset(8) + batchLength(4) + body
+                                if (cursor + 57 + 4 <= span.Length)
+                                    messageCount += System.Buffers.Binary.BinaryPrimitives.ReadInt32BigEndian(span.Slice(cursor + 57, 4));
+                                cursor += batchTotal;
+                            }
+
+                            partitionResponses.Add(new FetchResponse.PartitionResponse
+                            {
+                                Partition = partitionData.Partition,
+                                ErrorCode = ErrorCode.None,
+                                HighWatermark = highWatermark,
+                                RecordSet = recordSet,
+                            });
+                            continue;
+                        }
+                    }
+
                     if (!needsFiltering)
                     {
                         // Fast path: contiguous read — single allocation for all batches.

src/Kuestenlogik.Surgewave.Runtime/SurgewaveRuntime.cs

@@ -290,7 +290,7 @@ private async Task StartInternalAsync(CancellationToken cancellationToken)
         _metadataApiHandler = new MetadataApiHandler(config, _logManager, metadataApiLogger);
         IKafkaRequestHandler[] handlers =
         [
-            new DataApiHandler(config, _logManager, transactionCoordinator, _quotaManager, recordBatchSerializer, aclAuthorizer: null, deduplicationManager: null, delayIndex: null, ttlIndex: null, _metrics, dataApiLogger, partitionAppender: _options.PartitionAppender),
+            new DataApiHandler(config, _logManager, transactionCoordinator, _quotaManager, recordBatchSerializer, aclAuthorizer: null, deduplicationManager: null, delayIndex: null, ttlIndex: null, _metrics, dataApiLogger, partitionAppender: _options.PartitionAppender, disaggregatedReader: _options.DisaggregatedReader),
             _metadataApiHandler,
             _topicAdminHandler,
             new ConfigApiHandler(config, dynamicBrokerConfig, _logManager),

src/Kuestenlogik.Surgewave.Runtime/SurgewaveRuntimeBuilder.cs

@@ -1,5 +1,6 @@
 using Kuestenlogik.Surgewave.Broker;
 using Kuestenlogik.Surgewave.Core.Storage;
+using Kuestenlogik.Surgewave.Storage.Disaggregated.Read;
 using Kuestenlogik.Surgewave.Storage.Disaggregated.Routing;
 using Microsoft.Extensions.Logging;
 
@@ -46,6 +47,7 @@ public sealed class SurgewaveRuntimeBuilder
     private bool _cleanupOnDispose = true;
     private ILoggerFactory? _loggerFactory;
     private IPartitionAppender? _partitionAppender;
+    private DisaggregatedSegmentReader? _disaggregatedReader;
 
     internal SurgewaveRuntimeBuilder() { }
 
@@ -363,6 +365,20 @@ public SurgewaveRuntimeBuilder WithPartitionAppender(IPartitionAppender partitio
         return this;
     }
 
+    /// <summary>
+    /// Installs a disaggregated read fallback for the broker's Fetch path
+    /// (ADR-014, G21). When set, fetches against disaggregated topics
+    /// consult the partition manifest first and serve trimmed offsets from
+    /// the object store before falling back to the local WAL. Required
+    /// alongside <c>WalFlusherOptions.TrimAfterFlush</c> — without the
+    /// reader, trimmed offsets would be unreachable.
+    /// </summary>
+    public SurgewaveRuntimeBuilder WithDisaggregatedReader(DisaggregatedSegmentReader reader)
+    {
+        _disaggregatedReader = reader;
+        return this;
+    }
+
     // ==================== Configuration ====================
 
     /// <summary>
@@ -424,5 +440,6 @@ public SurgewaveRuntimeBuilder ConfigureFrom(BrokerConfig config)
         CleanupOnDispose = _cleanupOnDispose,
         LoggerFactory = _loggerFactory,
         PartitionAppender = _partitionAppender,
+        DisaggregatedReader = _disaggregatedReader,
     };
 }

src/Kuestenlogik.Surgewave.Runtime/SurgewaveRuntimeOptions.cs

@@ -1,5 +1,6 @@
 using Kuestenlogik.Surgewave.Broker;
 using Kuestenlogik.Surgewave.Core.Storage;
+using Kuestenlogik.Surgewave.Storage.Disaggregated.Read;
 using Kuestenlogik.Surgewave.Storage.Disaggregated.Routing;
 using Microsoft.Extensions.Logging;
 
@@ -198,6 +199,16 @@ public sealed record SurgewaveRuntimeOptions
     /// </summary>
     public IPartitionAppender? PartitionAppender { get; init; }
 
+    /// <summary>
+    /// Optional <see cref="DisaggregatedSegmentReader"/> that intercepts the
+    /// broker's Fetch read path. When the requested offset has been flushed
+    /// to the object store and the manifest knows the stream object key,
+    /// the reader hands back the bytes from S3 instead of hitting the local
+    /// WAL. Required for safe <c>WalFlusherOptions.TrimAfterFlush</c> usage.
+    /// Null = direct LogManager read (pre-G21 behaviour).
+    /// </summary>
+    public DisaggregatedSegmentReader? DisaggregatedReader { get; init; }
+
     // ==================== Internal ====================
 
     /// <summary>

src/Kuestenlogik.Surgewave.Storage.Disaggregated/Wal/PartitionLogWalSegmentSource.cs

@@ -70,15 +70,32 @@ public ValueTask<IReadOnlyList<WalSealedSegment>> ListSealedAsync(
             var lastOffset = s.CurrentOffset - 1;
             if (lastOffset < baseOffset) continue; // empty segment, nothing to flush
 
+            var logPath = Path.Combine(dir, $"{baseOffset:D20}.log");
+            var indexPath = Path.Combine(dir, $"{baseOffset:D20}.index");
+            var timeIndexPath = Path.Combine(dir, $"{baseOffset:D20}.timeindex");
             sealedSegments.Add(new WalSealedSegment(
                 Partition: partition,
                 BaseOffset: baseOffset,
                 LastOffset: lastOffset,
                 SizeBytes: s.Size,
                 CreatedAt: s.CreatedAt,
-                ReadLogBytesAsync: ct => ReadFileOrEmptyAsync(Path.Combine(dir, $"{baseOffset:D20}.log"), ct),
-                ReadIndexBytesAsync: ct => ReadFileOrEmptyAsync(Path.Combine(dir, $"{baseOffset:D20}.index"), ct),
-                ReadTimeIndexBytesAsync: ct => ReadFileOrEmptyAsync(Path.Combine(dir, $"{baseOffset:D20}.timeindex"), ct)));
+                ReadLogBytesAsync: ct => ReadFileOrEmptyAsync(logPath, ct),
+                ReadIndexBytesAsync: ct => ReadFileOrEmptyAsync(indexPath, ct),
+                ReadTimeIndexBytesAsync: ct => ReadFileOrEmptyAsync(timeIndexPath, ct))
+            {
+                TrimAsync = _ =>
+                {
+                    // Only delete the .log; the index siblings live with it
+                    // and we want to keep the trim atomic-ish from the
+                    // observer's POV. Missing files are tolerated — File.Delete
+                    // is idempotent. Errors propagate to the flusher's
+                    // try/catch which logs + swallows.
+                    DeleteIfExists(logPath);
+                    DeleteIfExists(indexPath);
+                    DeleteIfExists(timeIndexPath);
+                    return Task.CompletedTask;
+                },
+            });
         }
 
         return ValueTask.FromResult<IReadOnlyList<WalSealedSegment>>(sealedSegments);
@@ -93,4 +110,9 @@ private static async Task<byte[]> ReadFileOrEmptyAsync(string path, Cancellation
         if (!File.Exists(path)) return [];
         return await File.ReadAllBytesAsync(path, ct).ConfigureAwait(false);
     }
+
+    private static void DeleteIfExists(string path)
+    {
+        if (File.Exists(path)) File.Delete(path);
+    }
 }

src/Kuestenlogik.Surgewave.Storage.Disaggregated/Wal/WalFlusher.cs

@@ -142,5 +142,24 @@ await _remote.UploadSegmentAsync(
             CreatedAt: segment.CreatedAt);
 
         await _manifests.AppendObjectAsync(segment.Partition, manifestRef, cancellationToken).ConfigureAwait(false);
+
+        // Trim runs strictly AFTER the manifest commit succeeds. A failed
+        // upload or commit above throws and we never reach here, so the
+        // local segment file is preserved for the next scan to retry.
+        // Trim itself is allowed to fail (logged + swallowed): the
+        // manifest is the source of truth; a stray local file just costs
+        // disk until the retention sweeper picks it up.
+        if (_options.TrimAfterFlush && segment.TrimAsync is not null)
+        {
+            try
+            {
+                await segment.TrimAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex, "WAL-trim failed for {Topic}/{Partition} offset {Offset}; segment stays local",
+                    segment.Partition.Topic, segment.Partition.Partition, segment.BaseOffset);
+            }
+        }
     }
 }

src/Kuestenlogik.Surgewave.Storage.Disaggregated/Wal/WalFlusherOptions.cs

@@ -22,4 +22,15 @@ public sealed record WalFlusherOptions
     /// lot of catch-up to do. Default: 16.
     /// </summary>
     public int MaxSegmentsPerScan { get; init; } = 16;
+
+    /// <summary>
+    /// Whether to delete the local segment files after a successful flush
+    /// + manifest-commit. Off by default for safety: enabling this without
+    /// also wiring the disaggregated Fetch fallback
+    /// (<c>SurgewaveRuntimeBuilder.WithDisaggregatedReader</c>) would
+    /// make trimmed offsets unreachable. Turn on once the read fallback
+    /// is in place and you want the WAL footprint capped near zero
+    /// instead of following <c>retention.ms</c>.
+    /// </summary>
+    public bool TrimAfterFlush { get; init; }
 }

src/Kuestenlogik.Surgewave.Storage.Disaggregated/Wal/WalSealedSegment.cs

@@ -17,4 +17,23 @@ public sealed record WalSealedSegment(
     DateTime CreatedAt,
     Func<CancellationToken, Task<byte[]>> ReadLogBytesAsync,
     Func<CancellationToken, Task<byte[]>> ReadIndexBytesAsync,
-    Func<CancellationToken, Task<byte[]>> ReadTimeIndexBytesAsync);
+    Func<CancellationToken, Task<byte[]>> ReadTimeIndexBytesAsync)
+{
+    /// <summary>
+    /// Optional delegate that deletes the local segment files after a
+    /// successful flush + manifest-commit. When
+    /// <see cref="WalFlusherOptions.TrimAfterFlush"/> is on and this
+    /// delegate is set, the flusher invokes it as the final step of
+    /// <c>FlushOneAsync</c>. Null = keep the local copy (default; the
+    /// segment falls under the topic's normal <c>retention.ms</c> until
+    /// it expires).
+    ///
+    /// Safety note: the broker's Fetch path must consult the partition
+    /// manifest via <c>DisaggregatedSegmentReader</c> before reading
+    /// from the local log, otherwise a trimmed offset becomes
+    /// unreachable. The runtime wires this together when both
+    /// <c>WithPartitionAppender</c> and <c>WithDisaggregatedReader</c>
+    /// are set.
+    /// </summary>
+    public Func<CancellationToken, Task>? TrimAsync { get; init; }
+}

tests/Kuestenlogik.Surgewave.Storage.Disaggregated.Tests/Wal/WalFlusherTests.cs

@@ -116,6 +116,66 @@ public void StreamObjectKeyConvention_uses_D20_padding()
         Assert.Equal("topics/orders/0/stream-00000000000000000042.so", key);
     }
 
+    [Fact]
+    public async Task TrimAfterFlush_calls_TrimAsync_when_option_enabled()
+    {
+        var source = new FakeSource();
+        var trimmed = false;
+        source.Add(Segment(0, 99, 1024) with { TrimAsync = _ => { trimmed = true; return Task.CompletedTask; } });
+        var sut = NewFlusher(source, out _, out _,
+            options: new WalFlusherOptions { TrimAfterFlush = true });
+
+        await sut.RunOnceAsync(Partition);
+
+        Assert.True(trimmed);
+    }
+
+    [Fact]
+    public async Task TrimAfterFlush_off_skips_TrimAsync_even_when_delegate_set()
+    {
+        var source = new FakeSource();
+        var trimmed = false;
+        source.Add(Segment(0, 99, 1024) with { TrimAsync = _ => { trimmed = true; return Task.CompletedTask; } });
+        var sut = NewFlusher(source, out _, out _,
+            options: new WalFlusherOptions { TrimAfterFlush = false });
+
+        await sut.RunOnceAsync(Partition);
+
+        Assert.False(trimmed);
+    }
+
+    [Fact]
+    public async Task Failed_upload_skips_trim()
+    {
+        var source = new FakeSource();
+        var trimmed = false;
+        source.Add(Segment(0, 99, 1024) with { TrimAsync = _ => { trimmed = true; return Task.CompletedTask; } });
+        var sut = new WalFlusher(source, new InMemoryPartitionManifestStore(), new FailingRemote(),
+            options: new WalFlusherOptions { TrimAfterFlush = true });
+
+        await Assert.ThrowsAsync<IOException>(() => sut.RunOnceAsync(Partition));
+
+        Assert.False(trimmed);
+    }
+
+    [Fact]
+    public async Task Trim_exception_is_swallowed_after_successful_manifest_commit()
+    {
+        // A trim error after the manifest is already committed must not
+        // surface — the manifest is the source of truth, the leftover
+        // local file is a future janitor's problem.
+        var source = new FakeSource();
+        source.Add(Segment(0, 99, 1024) with { TrimAsync = _ => throw new IOException("disk full") });
+        var sut = NewFlusher(source, out _, out var manifests,
+            options: new WalFlusherOptions { TrimAfterFlush = true });
+
+        var n = await sut.RunOnceAsync(Partition);
+
+        Assert.Equal(1, n);
+        var manifest = await manifests.GetAsync(Partition);
+        Assert.Single(manifest.Objects);
+    }
+
     // ── fakes ────────────────────────────────────────────────────────────
 
     private static WalFlusher NewFlusher(