feat(storage): G21 P2a — PartitionManifestStore + 16 Tests

Erste konkrete P2-Bauteil: persistente Manifest-Layer fuer
disaggregated-Topics. KEIN Write-Path-Change, KEIN WAL-Flusher;
das ist die Datenhaltung auf der P2b/c aufbauen.

Neue Bestandteile:
- IPartitionManifestStore: GetAsync/AppendObjectAsync/ListPartitionsAsync.
  Optimistic-Concurrency via PartitionManifest.Version, Overlap-Check
  via PartitionManifest.AppendObject.
- InMemoryPartitionManifestStore: Dictionary + per-Partition-Semaphore.
  Verwendung in Tests + embedded Broker.
- FilePartitionManifestStore: JSON-Datei pro Partition unter
  data/disaggregated/manifests/<topic>__<partition>.json. Writes
  atomar via temp+rename. Hydration on-first-use, re-validiert
  Overlap-Invariante beim Boot.

Test-Coverage (16 Tests, alle passing):
- PartitionManifest: empty-state, append+version-bump, overlap-throw,
  contiguous-allowed, binary-search-Locate, RecordCount-inclusive.
- InMemory: round-trip, overlap-rejection, ListPartitions, concurrent-
  appends-to-different-partitions parallelisierbar.
- File: write-creates-file, persistence-across-recreation, no-tmp-
  residue, hydrate-from-disk, FileNameFor-convention.

InternalsVisibleTo fuer den Tests-Assembly, sonst Layer-Reinheit
bleibt — Disaggregated haengt nur an Core, keine Cycle Richtung Broker.

Author: thomas-stegemann

Kuestenlogik.Surgewave.slnx

@@ -30,6 +30,7 @@
     <Project Path="src/Kuestenlogik.Surgewave.Storage.Engine.Sqlite/Kuestenlogik.Surgewave.Storage.Engine.Sqlite.csproj" />
     <Project Path="src/Kuestenlogik.Surgewave.Storage.Tiering/Kuestenlogik.Surgewave.Storage.Tiering.csproj" />
     <Project Path="src/Kuestenlogik.Surgewave.Storage.Disaggregated/Kuestenlogik.Surgewave.Storage.Disaggregated.csproj" />
+    <Project Path="tests/Kuestenlogik.Surgewave.Storage.Disaggregated.Tests/Kuestenlogik.Surgewave.Storage.Disaggregated.Tests.csproj" />
     <Project Path="src/Kuestenlogik.Surgewave.Clustering/Kuestenlogik.Surgewave.Clustering.csproj" />
     <Project Path="src/Kuestenlogik.Surgewave.Broker/Kuestenlogik.Surgewave.Broker.csproj" />
     <Project Path="src/Kuestenlogik.Surgewave.Runtime/Kuestenlogik.Surgewave.Runtime.csproj" />

src/Kuestenlogik.Surgewave.Storage.Disaggregated/FilePartitionManifestStore.cs

@@ -0,0 +1,138 @@
+using System.Collections.Concurrent;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+using Kuestenlogik.Surgewave.Core.Models;
+
+namespace Kuestenlogik.Surgewave.Storage.Disaggregated;
+
+/// <summary>
+/// Manifest store that persists one JSON file per partition under a
+/// configured data directory. Writes are atomic (temp file + rename) so
+/// a crash mid-write never corrupts the manifest on disk — the broker
+/// either sees the old version or the new one, never a half-written
+/// file.
+///
+/// File layout under <c>&lt;dataDir&gt;/disaggregated/manifests/</c>:
+/// <code>
+///   manifests/
+///     orders__0.json
+///     orders__1.json
+///     events__0.json
+/// </code>
+/// The double-underscore is deliberate — Kafka topic names allow dots
+/// but not underscores in our convention, so <c>__</c> can never appear
+/// in a topic name and is therefore a safe partition separator.
+///
+/// All reads go through an in-memory cache so hot-path reads of the
+/// manifest don't touch disk; writes go to disk first, then update the
+/// cache.
+/// </summary>
+public sealed class FilePartitionManifestStore : IPartitionManifestStore
+{
+    private readonly string _root;
+    private readonly InMemoryPartitionManifestStore _cache = new();
+    private readonly ConcurrentDictionary<TopicPartition, SemaphoreSlim> _writeLocks = new();
+    private bool _hydrated;
+    private readonly SemaphoreSlim _hydrateLock = new(1, 1);
+
+    private static readonly JsonSerializerOptions JsonOptions = new()
+    {
+        WriteIndented = true,
+        DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
+    };
+
+    public FilePartitionManifestStore(string dataDirectory)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(dataDirectory);
+        _root = Path.Combine(Path.GetFullPath(dataDirectory), "disaggregated", "manifests");
+        Directory.CreateDirectory(_root);
+    }
+
+    public async ValueTask<PartitionManifest> GetAsync(TopicPartition partition, CancellationToken cancellationToken = default)
+    {
+        await EnsureHydratedAsync(cancellationToken).ConfigureAwait(false);
+        return await _cache.GetAsync(partition, cancellationToken).ConfigureAwait(false);
+    }
+
+    public async ValueTask<PartitionManifest> AppendObjectAsync(
+        TopicPartition partition,
+        StreamObjectRef newObject,
+        CancellationToken cancellationToken = default)
+    {
+        await EnsureHydratedAsync(cancellationToken).ConfigureAwait(false);
+
+        var gate = _writeLocks.GetOrAdd(partition, _ => new SemaphoreSlim(1, 1));
+        await gate.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            // Cache-Append performs the overlap check + version bump; that
+            // result is the manifest we serialise to disk. Disk is the
+            // durable side, but cache stays in lock-step so concurrent
+            // readers don't see a manifest that exists on disk but not
+            // in cache.
+            var next = await _cache.AppendObjectAsync(partition, newObject, cancellationToken).ConfigureAwait(false);
+            await WriteAtomicAsync(partition, next, cancellationToken).ConfigureAwait(false);
+            return next;
+        }
+        finally
+        {
+            gate.Release();
+        }
+    }
+
+    public async ValueTask<IReadOnlyList<TopicPartition>> ListPartitionsAsync(CancellationToken cancellationToken = default)
+    {
+        await EnsureHydratedAsync(cancellationToken).ConfigureAwait(false);
+        return await _cache.ListPartitionsAsync(cancellationToken).ConfigureAwait(false);
+    }
+
+    private async ValueTask EnsureHydratedAsync(CancellationToken cancellationToken)
+    {
+        if (Volatile.Read(ref _hydrated)) return;
+        await _hydrateLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            if (_hydrated) return;
+            foreach (var file in Directory.EnumerateFiles(_root, "*.json"))
+            {
+                var manifest = await ReadManifestAsync(file, cancellationToken).ConfigureAwait(false);
+                if (manifest is null) continue;
+
+                // Replay each ref through the in-memory store's Append so
+                // the overlap-check invariants are re-validated on every
+                // boot. A corrupted on-disk manifest fails loudly here
+                // instead of silently returning bad data on later reads.
+                foreach (var obj in manifest.Objects)
+                {
+                    await _cache.AppendObjectAsync(manifest.Partition, obj, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            Volatile.Write(ref _hydrated, true);
+        }
+        finally
+        {
+            _hydrateLock.Release();
+        }
+    }
+
+    private static async Task<PartitionManifest?> ReadManifestAsync(string path, CancellationToken cancellationToken)
+    {
+        await using var stream = File.OpenRead(path);
+        return await JsonSerializer.DeserializeAsync<PartitionManifest>(stream, JsonOptions, cancellationToken).ConfigureAwait(false);
+    }
+
+    private async Task WriteAtomicAsync(TopicPartition partition, PartitionManifest manifest, CancellationToken cancellationToken)
+    {
+        var finalPath = Path.Combine(_root, FileNameFor(partition));
+        var tempPath = finalPath + ".tmp";
+        await using (var stream = File.Create(tempPath))
+        {
+            await JsonSerializer.SerializeAsync(stream, manifest, JsonOptions, cancellationToken).ConfigureAwait(false);
+            await stream.FlushAsync(cancellationToken).ConfigureAwait(false);
+        }
+        File.Move(tempPath, finalPath, overwrite: true);
+    }
+
+    internal static string FileNameFor(TopicPartition partition) =>
+        $"{partition.Topic}__{partition.Partition}.json";
+}

src/Kuestenlogik.Surgewave.Storage.Disaggregated/IPartitionManifestStore.cs

@@ -0,0 +1,50 @@
+using Kuestenlogik.Surgewave.Core.Models;
+
+namespace Kuestenlogik.Surgewave.Storage.Disaggregated;
+
+/// <summary>
+/// Persistence layer for <see cref="PartitionManifest"/>. One manifest
+/// per disaggregated topic-partition; the store is the single source of
+/// truth for "what offset ranges live in the object store right now".
+///
+/// Two implementations ship in P2a:
+/// - <see cref="InMemoryPartitionManifestStore"/> for tests + embedded broker.
+/// - <see cref="FilePartitionManifestStore"/> for standalone brokers
+///   (atomic JSON write per partition). KRaft-replicated variant is a
+///   later iteration, see ADR-014 §"Per-partition manifest".
+///
+/// Implementations must be safe for concurrent calls across different
+/// partitions; same-partition mutations are serialised by the store so
+/// callers get optimistic-concurrency semantics via
+/// <see cref="PartitionManifest.Version"/> instead of needing external
+/// locks.
+/// </summary>
+public interface IPartitionManifestStore
+{
+    /// <summary>
+    /// Read the current manifest for <paramref name="partition"/>. Returns
+    /// an empty manifest (no stream objects, version 0) when the partition
+    /// has never been written to.
+    /// </summary>
+    ValueTask<PartitionManifest> GetAsync(TopicPartition partition, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Atomically append <paramref name="newObject"/> to the manifest. Bumps
+    /// <see cref="PartitionManifest.Version"/> on success and returns the
+    /// updated manifest. Throws <see cref="InvalidOperationException"/> if
+    /// the new object's offset range overlaps the manifest tail
+    /// (disaggregated commits must be strictly monotonic).
+    /// </summary>
+    ValueTask<PartitionManifest> AppendObjectAsync(
+        TopicPartition partition,
+        StreamObjectRef newObject,
+        CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// List every partition that has a manifest on this store. Used at
+    /// broker startup to rehydrate the in-memory manifest cache (and by
+    /// the cost-reporting endpoint to enumerate disaggregated storage
+    /// usage).
+    /// </summary>
+    ValueTask<IReadOnlyList<TopicPartition>> ListPartitionsAsync(CancellationToken cancellationToken = default);
+}

src/Kuestenlogik.Surgewave.Storage.Disaggregated/InMemoryPartitionManifestStore.cs

@@ -0,0 +1,54 @@
+using System.Collections.Concurrent;
+using Kuestenlogik.Surgewave.Core.Models;
+
+namespace Kuestenlogik.Surgewave.Storage.Disaggregated;
+
+/// <summary>
+/// In-process manifest store. Used in tests, in the embedded broker, and
+/// as the cache layer behind <see cref="FilePartitionManifestStore"/>. A
+/// <see cref="SemaphoreSlim"/> per partition serialises append calls
+/// without blocking unrelated partitions.
+/// </summary>
+public sealed class InMemoryPartitionManifestStore : IPartitionManifestStore
+{
+    private readonly ConcurrentDictionary<TopicPartition, PartitionManifest> _manifests = new();
+    private readonly ConcurrentDictionary<TopicPartition, SemaphoreSlim> _locks = new();
+
+    public ValueTask<PartitionManifest> GetAsync(TopicPartition partition, CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        var manifest = _manifests.TryGetValue(partition, out var existing)
+            ? existing
+            : PartitionManifest.Empty(partition);
+        return ValueTask.FromResult(manifest);
+    }
+
+    public async ValueTask<PartitionManifest> AppendObjectAsync(
+        TopicPartition partition,
+        StreamObjectRef newObject,
+        CancellationToken cancellationToken = default)
+    {
+        var gate = _locks.GetOrAdd(partition, _ => new SemaphoreSlim(1, 1));
+        await gate.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            var current = _manifests.TryGetValue(partition, out var existing)
+                ? existing
+                : PartitionManifest.Empty(partition);
+            var next = current.AppendObject(newObject);
+            _manifests[partition] = next;
+            return next;
+        }
+        finally
+        {
+            gate.Release();
+        }
+    }
+
+    public ValueTask<IReadOnlyList<TopicPartition>> ListPartitionsAsync(CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        IReadOnlyList<TopicPartition> snapshot = _manifests.Keys.ToArray();
+        return ValueTask.FromResult(snapshot);
+    }
+}

src/Kuestenlogik.Surgewave.Storage.Disaggregated/Kuestenlogik.Surgewave.Storage.Disaggregated.csproj

@@ -9,4 +9,8 @@
     <ProjectReference Include="..\Kuestenlogik.Surgewave.Core\Kuestenlogik.Surgewave.Core.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <InternalsVisibleTo Include="Kuestenlogik.Surgewave.Storage.Disaggregated.Tests" />
+  </ItemGroup>
+
 </Project>

tests/Kuestenlogik.Surgewave.Storage.Disaggregated.Tests/FilePartitionManifestStoreTests.cs

@@ -0,0 +1,89 @@
+using Kuestenlogik.Surgewave.Core.Models;
+using Kuestenlogik.Surgewave.Storage.Disaggregated;
+using Xunit;
+
+namespace Kuestenlogik.Surgewave.Storage.Disaggregated.Tests;
+
+/// <summary>
+/// File-backed store tests use a per-test temp dir so suites can run
+/// in parallel without colliding on disk.
+/// </summary>
+public sealed class FilePartitionManifestStoreTests : IDisposable
+{
+    private readonly string _tempDir;
+
+    public FilePartitionManifestStoreTests()
+    {
+        _tempDir = Path.Combine(Path.GetTempPath(), "swv-mfst-tests-" + Guid.NewGuid().ToString("N"));
+        Directory.CreateDirectory(_tempDir);
+    }
+
+    public void Dispose()
+    {
+        if (Directory.Exists(_tempDir)) Directory.Delete(_tempDir, recursive: true);
+    }
+
+    private static readonly TopicPartition P0 = new() { Topic = "orders", Partition = 0 };
+
+    [Fact]
+    public async Task Append_writes_json_file_at_expected_path()
+    {
+        var store = new FilePartitionManifestStore(_tempDir);
+        await store.AppendObjectAsync(P0, new StreamObjectRef("k", 0, 99, 1024, DateTime.UtcNow));
+
+        var expected = Path.Combine(_tempDir, "disaggregated", "manifests", "orders__0.json");
+        Assert.True(File.Exists(expected));
+    }
+
+    [Fact]
+    public async Task Manifest_survives_store_recreation()
+    {
+        var s1 = new FilePartitionManifestStore(_tempDir);
+        var first = new StreamObjectRef("k0", 0, 99, 1024, new DateTime(2026, 6, 11, 0, 0, 0, DateTimeKind.Utc));
+        var second = new StreamObjectRef("k1", 100, 199, 1024, new DateTime(2026, 6, 11, 0, 1, 0, DateTimeKind.Utc));
+        await s1.AppendObjectAsync(P0, first);
+        await s1.AppendObjectAsync(P0, second);
+
+        // Brand-new store instance — must re-hydrate from disk on first call.
+        var s2 = new FilePartitionManifestStore(_tempDir);
+        var reloaded = await s2.GetAsync(P0);
+
+        Assert.Equal(2, reloaded.Version);
+        Assert.Equal("k0", reloaded.Objects[0].ObjectKey);
+        Assert.Equal("k1", reloaded.Objects[1].ObjectKey);
+        Assert.Equal(0, reloaded.FirstOffset);
+        Assert.Equal(199, reloaded.LastOffset);
+    }
+
+    [Fact]
+    public async Task No_temp_file_remains_after_successful_append()
+    {
+        var store = new FilePartitionManifestStore(_tempDir);
+        await store.AppendObjectAsync(P0, new StreamObjectRef("k", 0, 99, 1024, DateTime.UtcNow));
+
+        var manifestDir = Path.Combine(_tempDir, "disaggregated", "manifests");
+        var lingering = Directory.GetFiles(manifestDir, "*.tmp");
+        Assert.Empty(lingering);
+    }
+
+    [Fact]
+    public async Task ListPartitions_returns_partitions_loaded_from_disk()
+    {
+        var s1 = new FilePartitionManifestStore(_tempDir);
+        await s1.AppendObjectAsync(P0, new StreamObjectRef("k", 0, 99, 1024, DateTime.UtcNow));
+        await s1.AppendObjectAsync(new TopicPartition { Topic = "orders", Partition = 1 },
+            new StreamObjectRef("k", 0, 99, 1024, DateTime.UtcNow));
+
+        var s2 = new FilePartitionManifestStore(_tempDir);
+        var partitions = await s2.ListPartitionsAsync();
+
+        Assert.Equal(2, partitions.Count);
+    }
+
+    [Fact]
+    public void FileNameFor_uses_double_underscore_separator()
+    {
+        var name = FilePartitionManifestStore.FileNameFor(new TopicPartition { Topic = "my.topic", Partition = 7 });
+        Assert.Equal("my.topic__7.json", name);
+    }
+}