From 0a77223b294ee550644724e4f00ab343ed57248b Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 10:58:36 +0100 Subject: [PATCH 01/10] feat(botinfra): aggregate snapshots for EventStore; vahter user streams use them Loading a user:* stream replayed every event on each message. Heavy streams (up to 3555 events, mostly frozen reaction history) cost 0.3-4s per load on cold heap pages (~1 page read per event). BotInfra.EventStore gains opt-in snapshotting (SnapshotPolicy): - state = stored snapshot + tail (`stream_version > snapshot`) in ONE statement - schema_version per aggregate; mismatching snapshots are ignored/overwritten - snapshots ahead of the log or undeserializable are deleted + replayed - monotonic upsert; write failures are recorded (metric + span) and swallowed, and on append run after commit, so the cache can never fail a write - request-scope cache holds snapshot-loaded state; any committed append invalidates both cached views of the stream - SnapshotShape.describe for pinning a state's shape to its schema version Vahter: V53 event_snapshot table, User.SnapshotPolicy (every 20 events), user appends + GetUserById go through snapshots; pin tests for User shape and Fold behaviour force a SchemaVersion bump on change. New tests/BotInfra.Tests (Testcontainers Postgres) wired into Vahter: Build. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/vahter-build.yml | 3 +- bots.slnx | 1 + src/BotInfra/BotInfra.fsproj | 1 + src/BotInfra/EVENTSTORE.md | 75 ++++- src/BotInfra/EventSnapshot.fs | 61 ++++ src/BotInfra/EventStore.fs | 307 +++++++++++++++++- src/VahterBanBot/DB.fs | 12 +- src/VahterBanBot/Types.fs | 4 + .../migrations/V53__event_snapshot.sql | 13 + tests/BotInfra.Tests/BotInfra.Tests.fsproj | 33 ++ .../BotInfra.Tests/EventStoreSnapshotTests.fs | 303 +++++++++++++++++ tests/BotInfra.Tests/PostgresFixture.fs | 52 +++ tests/BotInfra.Tests/Program.fs | 9 + tests/BotInfra.Tests/SnapshotShapeTests.fs | 41 +++ .../UserSnapshotTests.fs | 88 +++++ .../VahterBanBot.Unit.Tests.fsproj | 1 + 16 files changed, 986 insertions(+), 18 deletions(-) create mode 100644 src/BotInfra/EventSnapshot.fs create mode 100644 src/vahter-bot/migrations/V53__event_snapshot.sql create mode 100644 tests/BotInfra.Tests/BotInfra.Tests.fsproj create mode 100644 tests/BotInfra.Tests/EventStoreSnapshotTests.fs create mode 100644 tests/BotInfra.Tests/PostgresFixture.fs create mode 100644 tests/BotInfra.Tests/Program.fs create mode 100644 tests/BotInfra.Tests/SnapshotShapeTests.fs create mode 100644 tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs diff --git a/.github/workflows/vahter-build.yml b/.github/workflows/vahter-build.yml index 3776ad5..5ad1133 100644 --- a/.github/workflows/vahter-build.yml +++ b/.github/workflows/vahter-build.yml @@ -10,7 +10,8 @@ jobs: uses: ./.github/workflows/_bot-build.yml with: test-command: >- + dotnet test tests/BotInfra.Tests/BotInfra.Tests.fsproj -c Release --logger "trx;LogFileName=botinfra-test-results.trx" --logger "console;verbosity=detailed" && dotnet test tests/VahterBanBot.Unit.Tests/VahterBanBot.Unit.Tests.fsproj -c Release --logger "trx;LogFileName=unit-test-results.trx" --logger "console;verbosity=detailed" && dotnet test tests/SerializationCompat.Tests/SerializationCompat.Tests.fsproj -c Release --logger "trx;LogFileName=compat-test-results.trx" --logger "console;verbosity=detailed" && dotnet test tests/VahterBanBot.Tests/VahterBanBot.Tests.fsproj -c Release --logger "trx;LogFileName=test-results.trx" --logger "console;verbosity=detailed" - bot-paths-regex: '^(src/(BotInfra|VahterBanBot)/|src/Dockerfile\.bot$|tests/(VahterBanBot\.Tests|VahterBanBot\.Unit\.Tests|SerializationCompat\.Tests|BotTestInfra|FakeTgApi|FakeAzureOcrApi)/|Directory\.Build\.props$|Directory\.Packages\.props$|global\.json$)' + bot-paths-regex: '^(src/(BotInfra|VahterBanBot)/|src/Dockerfile\.bot$|tests/(BotInfra\.Tests|VahterBanBot\.Tests|VahterBanBot\.Unit\.Tests|SerializationCompat\.Tests|BotTestInfra|FakeTgApi|FakeAzureOcrApi)/|Directory\.Build\.props$|Directory\.Packages\.props$|global\.json$)' diff --git a/bots.slnx b/bots.slnx index 2af3c0c..9c08c5f 100644 --- a/bots.slnx +++ b/bots.slnx @@ -9,6 +9,7 @@ + diff --git a/src/BotInfra/BotInfra.fsproj b/src/BotInfra/BotInfra.fsproj index 68b03ec..98f86d9 100644 --- a/src/BotInfra/BotInfra.fsproj +++ b/src/BotInfra/BotInfra.fsproj @@ -8,6 +8,7 @@ + diff --git a/src/BotInfra/EVENTSTORE.md b/src/BotInfra/EVENTSTORE.md index 9e37dbb..d66984a 100644 --- a/src/BotInfra/EVENTSTORE.md +++ b/src/BotInfra/EVENTSTORE.md @@ -146,10 +146,81 @@ the projection callback always reflects the actual append version. If the decider returns `[]`, no events are written and the projection is not invoked — events and projection cannot drift. +## Snapshots (opt-in) + +Without a policy every load folds the entire stream. For long-lived aggregates +(vahter's `user:*` streams run into thousands of events), pass a snapshot table +and use the `SnapshotPolicy` overloads: state is loaded as *stored snapshot + +the events after it*, in one statement. + +```sql +CREATE TABLE event_snapshot ( + stream_id TEXT NOT NULL, + state_type TEXT NOT NULL, + schema_version INT NOT NULL, + stream_version INT NOT NULL, + state JSONB NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (stream_id, state_type) +); +GRANT SELECT, INSERT, UPDATE, DELETE ON event_snapshot TO ; +``` + +```fsharp +let store = EventStore(connString, "event", eventJsonOpts, "event_snapshot") + +type User = + ... + static member SnapshotPolicy = { StateType = "User"; SchemaVersion = 1; SnapshotEvery = 20 } + +let! (state, version) = EventStore.loadSnapshottedState store User.SnapshotPolicy streamId +let! (events, state) = EventStore.appendSnapshottedEventWithProjection store User.SnapshotPolicy streamId decider +// explicit fold/zero: store.LoadState / store.Transact / store.TransactWithProjection overloads +``` + +Rules the implementation guarantees: + +- **Snapshots are a cache, never the source of truth.** Truncating the table is + always safe; loads replay and re-create rows. A snapshot write failure is + recorded (`eventstore_snapshot_ops_total{op="write_failed"}` + span error) and + swallowed — it never fails a load, and on the append path it runs *after* the + commit so it can never roll back events. +- **Stale snapshots are detected and discarded.** A row whose `schema_version` + differs from the policy is ignored (and overwritten by the next write). A row + that is ahead of its log or doesn't deserialize is deleted and the stream is + replayed (`op="discarded_ahead" | "discarded_unreadable"`). +- **Snapshots only move forward** within a schema version (upsert guard), so a + slow writer can't regress a newer snapshot. A different schema version (new + code or a rollback) always overwrites. +- **Consistent reads.** Snapshot and tail come from one SQL statement, so a + concurrent snapshot rewrite can't make the tail skip events. Optimistic + concurrency is unchanged: expected version = snapshot version + tail length. +- **Write cadence.** A snapshot is written when a load or a committed append + leaves at least `SnapshotEvery` events past the stored one — frozen streams + get snapshotted lazily on their first slow load, no backfill needed. +- **Request scope.** `BeginRequestScope` caches snapshot-loaded states next to + raw streams; every committed append (through any API) drops both cached views + of that stream, then the appending path re-populates its own. + +### Versioning — the one thing you must not forget + +A snapshot bakes in two things: the state type's *shape* and the *result of +its fold*. Bump `SchemaVersion` on any change to either — a new field silently +deserializes to its default, and a fixed fold bug stays baked into old +snapshots. Enforce it with two pin tests next to the aggregate (see +`tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs`): + +1. `SnapshotShape.describe typeof` pinned per schema version. +2. The fold of a canonical event sequence (covering every case) pinned per + schema version. + +Both assert the *latest* pin's version equals the policy's, so changing the +type or the fold fails until the version is bumped and a new pin appended. +Also keep a JSON round-trip test and a "snapshot + tail = full replay at every +split point" test. + ## What this wrapper does NOT do -- **Snapshots / state caching.** Every read folds the entire stream. If you - have very long streams, project to a separate read model. - **Cross-stream transactions.** One stream per call. If you need to write to two streams atomically, model them as one stream. - **Schema migrations** of payload shapes. If you change a DU case, write a diff --git a/src/BotInfra/EventSnapshot.fs b/src/BotInfra/EventSnapshot.fs new file mode 100644 index 0000000..7765b38 --- /dev/null +++ b/src/BotInfra/EventSnapshot.fs @@ -0,0 +1,61 @@ +namespace BotInfra + +open System +open System.Collections.Generic +open Microsoft.FSharp.Reflection + +/// Opt-in `EventStore` snapshotting: state = stored snapshot + events after it. Snapshots are a +/// disposable cache — deleting them is always safe, loads fall back to a full replay. +type SnapshotPolicy = + { /// Aggregate discriminator, part of the snapshot key (one stream may carry several folds). + StateType: string + /// Must be bumped on ANY change to the state type's shape OR to its fold semantics; + /// snapshots written under another version are ignored and rebuilt by replay. + SchemaVersion: int + /// A snapshot is (re)written once at least this many events sit past the stored one. + SnapshotEvery: int } + +/// Deterministic structure of an F# type, for pinning a snapshotted state's shape to its +/// `SnapshotPolicy.SchemaVersion` in a test. Anonymous records render their fields only. +module SnapshotShape = + + let private isAnonymousRecord (t: Type) = t.Name.Contains "AnonymousType" + + let rec private render (visiting: HashSet) (t: Type) : string = + if t.IsGenericType && t.GetGenericTypeDefinition() = typedefof> then + $"option<{render visiting (t.GetGenericArguments()[0])}>" + elif t.IsGenericType && t.GetGenericTypeDefinition() = typedefof> then + $"list<{render visiting (t.GetGenericArguments()[0])}>" + elif t.IsGenericType && t.GetGenericTypeDefinition() = typedefof> then + $"nullable<{render visiting (t.GetGenericArguments()[0])}>" + elif t.IsArray then + $"array<{render visiting (t.GetElementType())}>" + elif FSharpType.IsTuple t then + FSharpType.GetTupleElements t |> Array.map (render visiting) |> String.concat " * " |> sprintf "(%s)" + elif FSharpType.IsRecord(t, true) || FSharpType.IsUnion(t, true) then + let name = if isAnonymousRecord t then "" else t.Name + if not (visiting.Add t) then $"@{name}" + else + let fields (ps: Reflection.PropertyInfo[]) = + ps |> Array.map (fun p -> $"{p.Name}: {render visiting p.PropertyType}") |> String.concat "; " + let body = + if FSharpType.IsRecord(t, true) then + $"{{{fields (FSharpType.GetRecordFields(t, true))}}}" + else + FSharpType.GetUnionCases(t, true) + |> Array.map (fun c -> + match c.GetFields() with + | [||] -> c.Name + | fs -> $"{c.Name} of ({fields fs})") + |> String.concat " | " + |> sprintf "[%s]" + %visiting.Remove t + name + body + elif t.IsEnum then + let names = Enum.GetNames t |> String.concat "," + $"{t.Name}<{names}>" + else + t.FullName + + /// Renders `t`'s structure; nested records/unions are expanded, recursion is cut with `@Name`. + let describe (t: Type) : string = render (HashSet()) t diff --git a/src/BotInfra/EventStore.fs b/src/BotInfra/EventStore.fs index fc776ba..cf8f355 100644 --- a/src/BotInfra/EventStore.fs +++ b/src/BotInfra/EventStore.fs @@ -26,6 +26,10 @@ module internal EventStoreTelemetry = meter.CreateCounter( "eventstore_stream_cache_mutations_total", "mutations", "Request-scoped stream-cache mutations, tagged by action (appended|evicted)") + let private snapshotOpsCounter = + meter.CreateCounter( + "eventstore_snapshot_ops_total", "ops", + "Aggregate-snapshot outcomes, tagged by state_type and op (hit|miss|written|discarded_ahead|discarded_unreadable|write_failed)") /// Records a load and tags the (possibly null) load span with its provenance. let recordLoad (activity: Activity) (source: string) (streamId: string) (version: int) (eventCount: int) = @@ -45,6 +49,12 @@ module internal EventStoreTelemetry = | null -> () | a -> a.SetTag("eventstore.cache.action", action) |> ignore + /// Records a snapshot outcome on a metric and as a tag on the (possibly null) span. + let recordSnapshot (activity: Activity) (stateType: string) (op: string) = + snapshotOpsCounter.Add(1L, KeyValuePair("state_type", box stateType), KeyValuePair("op", box op)) + if not (isNull activity) then + activity.AddEvent(ActivityEvent($"snapshot.{op}")) |> ignore + /// Raw row materialized from the per-bot event table. /// `data` is JSONB read back as TEXT so Dapper can map it as a plain string. [] @@ -60,6 +70,28 @@ type RawEvent = /// expected version. The caller's job is to re-read state and retry. type ConcurrencyConflict = ConcurrencyConflict +/// A snapshot-aware load row: the usable snapshot (if any) first, then the tail events. +/// `head_version` (snapshot row only) is the log's max version, to detect a snapshot ahead of it. +[] +type internal SnapshotLoadRow = + { is_snapshot: bool + stream_version: int + data: string + head_version: Nullable } + +/// Snapshot-loaded aggregate state; `SnapshotVersion` is the version persisted in the snapshot table. +type internal LoadedState = + { StateType: string + State: obj + Version: int + SnapshotVersion: int } + +/// Request-scoped identity map: raw event lists and snapshot-loaded states, keyed by stream id. +[] +type internal RequestCache() = + member val Raws = Dictionary() + member val States = Dictionary() + /// Append-only event store wrapper. One instance per (connection-string, event-table) /// pair. Each bot owns its own event table — this wrapper does not attempt to merge them. /// @@ -73,13 +105,19 @@ type ConcurrencyConflict = ConcurrencyConflict /// with `WithUnionInternalTag()` is the canonical pick. /// /// The event table must follow the schema described in `EVENTSTORE.md`. -type EventStore(connString: string, tableName: string, jsonOptions: JsonSerializerOptions) = +/// `snapshotTableName` (optional, same schema doc) enables the `SnapshotPolicy` overloads. +type EventStore(connString: string, tableName: string, jsonOptions: JsonSerializerOptions, ?snapshotTableName: string) = + static let identifierPattern = @"^[a-z_][a-z0-9_]{0,62}$" do if isNull connString then nullArg (nameof connString) if isNull tableName then nullArg (nameof tableName) if isNull jsonOptions then nullArg (nameof jsonOptions) - if not (Regex.IsMatch(tableName, @"^[a-z_][a-z0-9_]{0,62}$")) then + if not (Regex.IsMatch(tableName, identifierPattern)) then invalidArg (nameof tableName) $"invalid event table name: %s{tableName}" + match snapshotTableName with + | Some snap when not (Regex.IsMatch(snap, identifierPattern)) -> + invalidArg (nameof snapshotTableName) $"invalid snapshot table name: %s{snap}" + | _ -> () let selectAllSql = $""" @@ -100,6 +138,58 @@ RETURNING id let maxVersionSql = $"SELECT MAX(stream_version) FROM {tableName} WHERE stream_id = @streamId" + let snapshotTable = defaultArg snapshotTableName "" + + let validatePolicy (policy: SnapshotPolicy) = + if snapshotTableName.IsNone then + invalidOp "SnapshotPolicy requires an EventStore constructed with snapshotTableName" + if String.IsNullOrWhiteSpace policy.StateType then + invalidArg (nameof policy) "SnapshotPolicy.StateType must be set" + if policy.SnapshotEvery < 1 then + invalidArg (nameof policy) "SnapshotPolicy.SnapshotEvery must be >= 1" + + // One statement, so the snapshot and the tail boundary come from the same MVCC snapshot — + // a concurrent snapshot rewrite can never make the tail skip events. + let snapshotLoadSql = + $""" +WITH snap AS ( + SELECT stream_version, state + FROM {snapshotTable} + WHERE stream_id = @streamId AND state_type = @stateType AND schema_version = @schemaVersion +) +SELECT TRUE AS is_snapshot, s.stream_version, s.state::TEXT AS data, + (SELECT MAX(stream_version) FROM {tableName} WHERE stream_id = @streamId) AS head_version +FROM snap s +UNION ALL +SELECT FALSE, e.stream_version, e.data::TEXT, NULL +FROM {tableName} e +WHERE e.stream_id = @streamId + AND e.stream_version > COALESCE((SELECT stream_version FROM snap), 0) +ORDER BY is_snapshot DESC, stream_version +""" + + // Never moves a snapshot backwards within a schema version; a different schema version + // (newer code, or a rollback) always overwrites. + let snapshotWriteSql = + $""" +INSERT INTO {snapshotTable} (stream_id, state_type, schema_version, stream_version, state) +VALUES (@streamId, @stateType, @schemaVersion, @streamVersion, @state::JSONB) +ON CONFLICT (stream_id, state_type) DO UPDATE + SET schema_version = EXCLUDED.schema_version, + stream_version = EXCLUDED.stream_version, + state = EXCLUDED.state, + updated_at = now() + WHERE {snapshotTable}.schema_version <> EXCLUDED.schema_version + OR {snapshotTable}.stream_version < EXCLUDED.stream_version +""" + + let snapshotDeleteSql = + $""" +DELETE FROM {snapshotTable} +WHERE stream_id = @streamId AND state_type = @stateType + AND schema_version = @schemaVersion AND stream_version = @streamVersion +""" + // Request-scoped identity map (unit of work): while a scope is active, repeated loads of the // same stream within one handle are served from memory instead of re-querying Postgres. The // AsyncLocal value is null outside a scope (no caching). Only the `readStream` path is cached; @@ -108,24 +198,50 @@ RETURNING id // NOTE: cached raws are only ever folded by `data` + `stream_version` (id/created_at/event_type // are never read off the cache — verified for all readStream consumers), so appended events are // synthesized in-memory below without a re-read. - let scopedCache = AsyncLocal>() + let scopedCache = AsyncLocal() let cacheTryGet (streamId: string) : (RawEvent list * int) voption = let c = scopedCache.Value if isNull c then ValueNone else - match c.TryGetValue streamId with + match c.Raws.TryGetValue streamId with | true, v -> ValueSome v | _ -> ValueNone let cachePut (streamId: string) (entry: RawEvent list * int) = let c = scopedCache.Value - if not (isNull c) then c[streamId] <- entry + if not (isNull c) then c.Raws[streamId] <- entry let cacheEvict (streamId: string) = let c = scopedCache.Value if not (isNull c) then - if c.Remove streamId then EventStoreTelemetry.recordMutation "evicted" streamId + if c.Raws.Remove streamId then EventStoreTelemetry.recordMutation "evicted" streamId + + // Snapshot-loaded states live in the same scope; every committed append drops both views. + let stateCacheTryGet (streamId: string) (stateType: string) : LoadedState option = + let c = scopedCache.Value + if isNull c then None + else + match c.States.TryGetValue streamId with + | true, e when e.StateType = stateType -> Some e + | _ -> None + + let stateCachePut (streamId: string) (entry: LoadedState) = + let c = scopedCache.Value + if not (isNull c) then c.States[streamId] <- entry + + let stateCacheEvict (streamId: string) = + let c = scopedCache.Value + if not (isNull c) then + if c.States.Remove streamId then EventStoreTelemetry.recordMutation "evicted" streamId + + /// Drops every cached view of a stream after a committed append; the caller re-populates + /// whichever view it maintains, so no other view can serve pre-append data. + let invalidateStream (streamId: string) = + let c = scopedCache.Value + if not (isNull c) then + %c.Raws.Remove streamId + %c.States.Remove streamId /// Reflects an append into the cache (if a scope is active) by synthesizing the new rows in /// memory, so a subsequent load in the same handle is free and reflects our own write. @@ -143,9 +259,106 @@ RETURNING id data = JsonSerializer.Serialize<'TEvent>(e, jsonOptions) created_at = Unchecked.defaultof }) let newVersion = baseVersion + List.length newEvents - c[streamId] <- (priorRaws @ synthesized, newVersion) + c.Raws[streamId] <- (priorRaws @ synthesized, newVersion) EventStoreTelemetry.recordMutation "appended" streamId + /// Upserts a snapshot. A failure is recorded (metric + span error) and swallowed: snapshots + /// are a cache, so a failed write must never fail the load or the committed append. + let tryWriteSnapshot (conn: NpgsqlConnection) (activity: Activity) (policy: SnapshotPolicy) + (streamId: string) (version: int) (state: 'State) : Task = + task { + try + let json = JsonSerializer.Serialize<'State>(state, jsonOptions) + let! _ = + conn.ExecuteAsync(snapshotWriteSql, + {| streamId = streamId; stateType = policy.StateType; schemaVersion = policy.SchemaVersion + streamVersion = version; state = json |}) + EventStoreTelemetry.recordSnapshot activity policy.StateType "written" + return true + with ex -> + EventStoreTelemetry.recordSnapshot activity policy.StateType "write_failed" + if not (isNull activity) then + %activity.AddException(ex).SetStatus(ActivityStatusCode.Error, "snapshot write failed") + return false + } + + /// Loads aggregate state as snapshot + tail. A snapshot that is ahead of its log or can't be + /// deserialized is deleted and the stream is replayed from scratch. + let loadSnapshotted (fold: 'State -> 'TEvent -> 'State) (zero: 'State) (policy: SnapshotPolicy) + (streamId: string) : Task = + task { + validatePolicy policy + match stateCacheTryGet streamId policy.StateType with + | Some entry -> + use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.load") + EventStoreTelemetry.recordLoad activity "cache" streamId entry.Version 0 + return entry + | None -> + use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.load") + use conn = new NpgsqlConnection(connString) + let keyArgs = + {| streamId = streamId; stateType = policy.StateType; schemaVersion = policy.SchemaVersion |} + let! rows = conn.QueryAsync(snapshotLoadSql, keyArgs) + let rows = List.ofSeq rows + let snapshotRow, tailRows = + match rows with + | r :: rest when r.is_snapshot -> Some r, rest + | _ -> None, rows + let tail = tailRows |> List.map (fun r -> r.stream_version, r.data) + let fromSnapshot = + match snapshotRow with + | None -> Ok None + | Some r when not r.head_version.HasValue || r.head_version.Value < r.stream_version -> + Error ("discarded_ahead", r.stream_version) + | Some r -> + try + match box (JsonSerializer.Deserialize<'State>(r.data, jsonOptions)) with + | null -> Error ("discarded_unreadable", r.stream_version) + | s -> Ok (Some (unbox<'State> s, r.stream_version)) + with ex -> + if not (isNull activity) then %activity.AddException ex + Error ("discarded_unreadable", r.stream_version) + let! (baseState, baseVersion, events) = + task { + match fromSnapshot with + | Ok (Some (s, v)) -> + EventStoreTelemetry.recordSnapshot activity policy.StateType "hit" + return s, v, tail + | Ok None -> + EventStoreTelemetry.recordSnapshot activity policy.StateType "miss" + return zero, 0, tail + | Error (op, badVersion) -> + EventStoreTelemetry.recordSnapshot activity policy.StateType op + let! _ = + conn.ExecuteAsync(snapshotDeleteSql, + {| streamId = streamId; stateType = policy.StateType + schemaVersion = policy.SchemaVersion; streamVersion = badVersion |}) + let! raws = conn.QueryAsync(selectAllSql, {| streamId = streamId |}) + return zero, 0, (raws |> Seq.map (fun r -> r.stream_version, r.data) |> List.ofSeq) + } + let state = + events + |> List.fold (fun s (_, data) -> fold s (JsonSerializer.Deserialize<'TEvent>(data, jsonOptions))) baseState + let version = + match List.tryLast events with + | Some (v, _) -> v + | None -> baseVersion + let! snapshotVersion = + if version - baseVersion >= policy.SnapshotEvery then + task { + let! written = tryWriteSnapshot conn activity policy streamId version state + return if written then version else baseVersion + } + else Task.FromResult baseVersion + EventStoreTelemetry.recordLoad activity "db" streamId version (List.length events) + if not (isNull activity) then %activity.SetTag("snapshot_version", baseVersion) + let entry = + { StateType = policy.StateType; State = box state + Version = version; SnapshotVersion = snapshotVersion } + stateCachePut streamId entry + return entry + } + let readStream (streamId: string) : Task = task { match cacheTryGet streamId with @@ -189,7 +402,7 @@ RETURNING id /// Intended to wrap one update handle — see the webhook entry point. member _.BeginRequestScope() : IDisposable = let prev = scopedCache.Value - scopedCache.Value <- Dictionary() + scopedCache.Value <- RequestCache() { new IDisposable with member _.Dispose() = scopedCache.Value <- prev } /// Returns the highest stream_version for the given stream, or 0 if the stream is empty. @@ -285,6 +498,7 @@ ORDER BY stream_id, stream_version return Error ConcurrencyConflict else do! tx.CommitAsync() + invalidateStream streamId return Ok () } @@ -310,6 +524,7 @@ ORDER BY stream_id, stream_version else do! projection conn tx do! tx.CommitAsync() + invalidateStream streamId return Ok () } @@ -384,6 +599,64 @@ ORDER BY stream_id, stream_version return result.Value } + /// Loads state as snapshot + tail with the stream version (`0` = no stream yet), refreshing + /// the snapshot once at least `policy.SnapshotEvery` events were folded past it. + member _.LoadState<'TEvent, 'State> + (fold: 'State -> 'TEvent -> 'State, zero: 'State, policy: SnapshotPolicy, streamId: string) + : Task<'State * int> = + task { + let! loaded = loadSnapshotted fold zero policy streamId + return unbox<'State> loaded.State, loaded.Version + } + + /// Snapshot-aware `TransactWithProjection`. The snapshot refresh runs after the commit, so a + /// failing snapshot write can never roll back the appended events. + member this.TransactWithProjection<'TEvent, 'State> + (fold: 'State -> 'TEvent -> 'State, zero: 'State, policy: SnapshotPolicy, + decider: 'State -> 'TEvent list * (NpgsqlConnection -> NpgsqlTransaction -> Task) option, + streamId: string) + : Task<'TEvent list * 'State> = + task { + let mutable result = None + while result.IsNone do + let! loaded = loadSnapshotted fold zero policy streamId + let state = unbox<'State> loaded.State + let (newEvents, projection) = decider state + if newEvents.IsEmpty then + result <- Some ([], state) + else + let proj = + match projection with + | Some p -> p + | None -> fun _ _ -> Task.CompletedTask + match! this.TryAppendWithProjection(streamId, loaded.Version, newEvents, proj) with + | Ok _ -> + let finalState = newEvents |> List.fold fold state + let newVersion = loaded.Version + newEvents.Length + let! snapshotVersion = + if newVersion - loaded.SnapshotVersion >= policy.SnapshotEvery then + task { + use conn = new NpgsqlConnection(connString) + let! written = tryWriteSnapshot conn Activity.Current policy streamId newVersion finalState + return if written then newVersion else loaded.SnapshotVersion + } + else Task.FromResult loaded.SnapshotVersion + stateCachePut streamId + { loaded with State = box finalState; Version = newVersion; SnapshotVersion = snapshotVersion } + result <- Some (newEvents, finalState) + | Error ConcurrencyConflict -> + // Stale read lost the version race — the retry must re-read from the DB. + stateCacheEvict streamId + return result.Value + } + + /// Snapshot-aware `Transact` (no projection). + member this.Transact<'TEvent, 'State> + (fold: 'State -> 'TEvent -> 'State, zero: 'State, policy: SnapshotPolicy, + decider: 'State -> 'TEvent list, streamId: string) + : Task<'TEvent list * 'State> = + this.TransactWithProjection(fold, zero, policy, (fun s -> decider s, None), streamId) + /// Companion module — SRTP convenience that resolves Fold/Zero from the state type /// at compile time, so callers don't have to thread them through every callsite. [] @@ -416,3 +689,21 @@ module EventStore = and 'State : (static member Fold : 'State * 'TEvent -> 'State) = let fold s e = 'State.Fold(s, e) store.TransactWithProjection(fold, 'State.Zero, decider, streamId) + + /// SRTP wrapper for `store.LoadState`; type arguments are explicit because `'TEvent` can't be inferred. + let inline loadSnapshottedState<'TEvent, 'State + when 'State : (static member Zero : 'State) + and 'State : (static member Fold : 'State * 'TEvent -> 'State)> + (store: EventStore) (policy: SnapshotPolicy) (streamId: string) : Task<'State * int> = + let fold s e = 'State.Fold(s, e) + store.LoadState<'TEvent, 'State>(fold, 'State.Zero, policy, streamId) + + /// Snapshot-aware `appendEventWithProjection`. + let inline appendSnapshottedEventWithProjection + (store: EventStore) (policy: SnapshotPolicy) (streamId: string) + (decider: 'State -> 'TEvent list * (NpgsqlConnection -> NpgsqlTransaction -> Task) option) + : Task<'TEvent list * 'State> + when 'State : (static member Zero : 'State) + and 'State : (static member Fold : 'State * 'TEvent -> 'State) = + let fold s e = 'State.Fold(s, e) + store.TransactWithProjection(fold, 'State.Zero, policy, decider, streamId) diff --git a/src/VahterBanBot/DB.fs b/src/VahterBanBot/DB.fs index c8200f3..4e799ce 100644 --- a/src/VahterBanBot/DB.fs +++ b/src/VahterBanBot/DB.fs @@ -66,7 +66,7 @@ type SpamOrHamDb = type DbService(connString: string, timeProvider: TimeProvider) = let utcNow () = timeProvider.GetUtcNow().UtcDateTime - let store = EventStore(connString, "event", eventJsonOpts) + let store = EventStore(connString, "event", eventJsonOpts, "event_snapshot") // ----------------------------------------------------------------------- // Snapshot read-model upserts (run in the SAME TX as the event append). @@ -153,7 +153,7 @@ UPDATE snapshot_message // Append wrappers: fold the new events to final state, serialize the snapshot DTO, // and attach the matching upsert as the in-TX projection. let appendUserEvents (userId: int64) (decide: User -> UserEvent list) : Task = - EventStore.appendEventWithProjection store $"user:{userId}" (fun (state: User) -> + EventStore.appendSnapshottedEventWithProjection store User.SnapshotPolicy $"user:{userId}" (fun (state: User) -> match decide state with | [] -> [], None | evts -> @@ -323,11 +323,9 @@ ON CONFLICT DO NOTHING member _.GetUserById(userId: int64) : Task = task { - let! events = store.GetEventsForStream($"user:{userId}") - if events.Length = 0 then return None - else - let state = (User.Zero, events) ||> Array.fold (fun s e -> User.Fold(s, e)) - return Some { state with Id = userId } + let! (state, version) = EventStore.loadSnapshottedState store User.SnapshotPolicy $"user:{userId}" + if version = 0 then return None + else return Some { state with Id = userId } } // ----------------------------------------------------------------------- diff --git a/src/VahterBanBot/Types.fs b/src/VahterBanBot/Types.fs index 165878e..d284ce2 100644 --- a/src/VahterBanBot/Types.fs +++ b/src/VahterBanBot/Types.fs @@ -168,6 +168,10 @@ type User = static member Zero = { Id = 0L; Banned = None; Username = None; ReactionCount = 0; NotSpamUntil = None SpamProtectionUntil = None; SpamProtectionHits = 0 } + /// SchemaVersion MUST be bumped on any change to User's fields or to Fold's behaviour — + /// UserSnapshotTests pins both and fails until it is. + static member SnapshotPolicy : SnapshotPolicy = + { StateType = "User"; SchemaVersion = 1; SnapshotEvery = 20 } static member Fold (state: User, event: UserEvent) : User = match event with | UsernameChanged e -> { state with Id = e.userId; Username = e.username } diff --git a/src/vahter-bot/migrations/V53__event_snapshot.sql b/src/vahter-bot/migrations/V53__event_snapshot.sql new file mode 100644 index 0000000..36a9276 --- /dev/null +++ b/src/vahter-bot/migrations/V53__event_snapshot.sql @@ -0,0 +1,13 @@ +-- Aggregate snapshots for BotInfra.EventStore (see src/BotInfra/EVENTSTORE.md). A disposable +-- cache: truncating it is always safe, loads fall back to a full stream replay. +CREATE TABLE IF NOT EXISTS event_snapshot ( + stream_id TEXT NOT NULL, + state_type TEXT NOT NULL, + schema_version INT NOT NULL, + stream_version INT NOT NULL, + state JSONB NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (stream_id, state_type) +); + +GRANT SELECT, INSERT, UPDATE, DELETE ON event_snapshot TO vahter_bot_ban_service; diff --git a/tests/BotInfra.Tests/BotInfra.Tests.fsproj b/tests/BotInfra.Tests/BotInfra.Tests.fsproj new file mode 100644 index 0000000..0c06557 --- /dev/null +++ b/tests/BotInfra.Tests/BotInfra.Tests.fsproj @@ -0,0 +1,33 @@ + + + + Exe + false + false + true + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs new file mode 100644 index 0000000..95559c1 --- /dev/null +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -0,0 +1,303 @@ +module BotInfra.Tests.EventStoreSnapshotTests + +open System +open System.Collections.Concurrent +open System.Diagnostics +open System.Text.Json.Serialization +open System.Threading.Tasks +open Dapper +open Npgsql +open BotInfra +open BotInfra.Tests.PostgresFixture +open Xunit + +type TallyEvent = + | Added of {| amount: int |} + | Renamed of {| name: string option |} + +type Tally = + { Total: int + Name: string option + Count: int } + static member Zero = { Total = 0; Name = None; Count = 0 } + static member Fold(s: Tally, e: TallyEvent) = + match e with + | Added a -> { s with Total = s.Total + a.amount; Count = s.Count + 1 } + | Renamed r -> { s with Name = r.name; Count = s.Count + 1 } + +let private fold s e = Tally.Fold(s, e) + +let private jsonOpts = + JsonFSharpOptions.Default() + .WithUnionInternalTag() + .WithUnionUnwrapRecordCases() + .WithUnionNamedFields() + .WithUnwrapOption() + .WithSkippableOptionFields(SkippableOptionFields.Always, deserializeNullAsNone = true) + .ToJsonSerializerOptions() + +let private policy every = { StateType = "Tally"; SchemaVersion = 1; SnapshotEvery = every } + +/// Every `eventStore.load` span stopped while the capture is alive. +type private LoadSpans() = + let spans = ConcurrentQueue() + let listener = + new ActivityListener( + ShouldListenTo = (fun s -> s.Name = "BotInfra.EventStore"), + Sample = (fun _ -> ActivitySamplingResult.AllDataAndRecorded), + ActivityStopped = (fun a -> if a.OperationName = "eventStore.load" then spans.Enqueue a)) + do ActivitySource.AddActivityListener listener + member _.All = List.ofSeq spans + member _.Last = Seq.last spans + interface IDisposable with + member _.Dispose() = listener.Dispose() + +let private tag (key: string) (a: Activity) = a.GetTagItem key |> string +let private hasEvent (name: string) (a: Activity) = a.Events |> Seq.exists (fun e -> e.Name = name) + +type SnapshotRow = { schema_version: int; stream_version: int; state: string } + +type EventStoreSnapshotTests(db: PostgresFixture) = + let store = EventStore(db.ConnectionString, "event", jsonOpts, "event_snapshot") + let newStream () = $"tally:{Guid.NewGuid():N}" + + let appendRaw (streamId: string) (events: TallyEvent list) = task { + let! (_, version) = store.GetRawEventsForStream streamId + match! store.TryAppend(streamId, version, events) with + | Ok () -> () + | Error _ -> failwith "unexpected concurrency conflict" + } + + let added n = [ for i in 1 .. n -> Added {| amount = i |} ] + + let replay streamId = store.FoldEvents(fold, Tally.Zero, streamId) + + let snapshotRow (table: string) (streamId: string) = task { + use conn = new NpgsqlConnection(db.ConnectionString) + let! rows = + conn.QueryAsync( + $"SELECT schema_version, stream_version, state::TEXT AS state FROM {table} WHERE stream_id = @streamId", + {| streamId = streamId |}) + return Seq.tryHead rows + } + + let exec (sql: string) (streamId: string) = task { + use conn = new NpgsqlConnection(db.ConnectionString) + let! _ = conn.ExecuteAsync(sql, {| streamId = streamId |}) + return () + } + + let load p streamId = store.LoadState(fold, Tally.Zero, p, streamId) + + let increment p streamId amount = + store.Transact(fold, Tally.Zero, p, (fun _ -> [ Added {| amount = amount |} ]), streamId) + + [] + member _.``first load replays the stream and snapshots it once the tail reaches SnapshotEvery``() = task { + let sid = newStream () + do! appendRaw sid (added 5) + let! (state, version) = load (policy 3) sid + let! expected = replay sid + Assert.Equal(expected, state) + Assert.Equal(5, version) + let! row = snapshotRow "event_snapshot" sid + Assert.Equal(Some 5, row |> Option.map _.stream_version) + + use spans = new LoadSpans() + let! (again, _) = load (policy 3) sid + Assert.Equal(expected, again) + Assert.Equal("0", tag "event_count" spans.Last) + Assert.True(hasEvent "snapshot.hit" spans.Last) + } + + [] + member _.``no snapshot is written while the tail is below SnapshotEvery``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + let! (state, version) = load (policy 3) sid + Assert.Equal(3, state.Total) + Assert.Equal(2, version) + let! row = snapshotRow "event_snapshot" sid + Assert.Equal(None, row) + } + + [] + member _.``missing stream loads as zero at version 0``() = task { + let! (state, version) = load (policy 1) (newStream ()) + Assert.Equal(Tally.Zero, state) + Assert.Equal(0, version) + } + + [] + member _.``a stored snapshot is used instead of replaying the prefix``() = task { + let sid = newStream () + do! appendRaw sid (added 3) + let! _ = load (policy 1) sid + do! exec """UPDATE event_snapshot SET state = jsonb_set(state, '{Total}', '1000') WHERE stream_id = @streamId""" sid + do! appendRaw sid [ Added {| amount = 1 |} ] + let! (state, version) = load (policy 100) sid + Assert.Equal(1001, state.Total) + Assert.Equal(4, version) + } + + [] + member _.``snapshot plus tail equals full replay across snapshot boundaries``() = task { + let sid = newStream () + let p = policy 3 + for i in 1 .. 10 do + let! _ = + store.Transact(fold, Tally.Zero, p, + (fun (s: Tally) -> + if i % 4 = 0 then [ Renamed {| name = Some $"n{s.Total}" |} ] + else [ Added {| amount = i |} ]), + sid) + let! (loaded, version) = load p sid + let! expected = replay sid + Assert.Equal(expected, loaded) + Assert.Equal(i, version) + let! row = snapshotRow "event_snapshot" sid + Assert.Equal(Some 9, row |> Option.map _.stream_version) + } + + [] + member _.``a snapshot from another schema version is ignored and overwritten``() = task { + let sid = newStream () + do! appendRaw sid (added 4) + let! _ = load (policy 1) sid + do! exec """UPDATE event_snapshot SET state = jsonb_set(state, '{Total}', '1000') WHERE stream_id = @streamId""" sid + + use spans = new LoadSpans() + let v2 = { policy 1 with SchemaVersion = 2 } + let! (state, _) = load v2 sid + Assert.Equal(10, state.Total) + Assert.True(hasEvent "snapshot.miss" spans.Last) + Assert.Equal("4", tag "event_count" spans.Last) + let! row = snapshotRow "event_snapshot" sid + Assert.Equal(Some 2, row |> Option.map _.schema_version) + Assert.DoesNotContain("1000", row.Value.state) + } + + [] + member _.``a snapshot ahead of its log is discarded and rebuilt``() = task { + let sid = newStream () + do! appendRaw sid (added 4) + let! _ = load (policy 1) sid + do! exec "UPDATE event_snapshot SET stream_version = 999 WHERE stream_id = @streamId" sid + + use spans = new LoadSpans() + let! (state, version) = load (policy 1) sid + Assert.Equal(10, state.Total) + Assert.Equal(4, version) + Assert.True(hasEvent "snapshot.discarded_ahead" spans.Last) + let! row = snapshotRow "event_snapshot" sid + Assert.Equal(Some 4, row |> Option.map _.stream_version) + } + + [] + member _.``an unreadable snapshot is discarded and rebuilt``() = task { + let sid = newStream () + do! appendRaw sid (added 4) + let! _ = load (policy 1) sid + do! exec """UPDATE event_snapshot SET state = '{"unexpected": true}' WHERE stream_id = @streamId""" sid + + use spans = new LoadSpans() + let! (state, _) = load (policy 1) sid + Assert.Equal(10, state.Total) + Assert.True(hasEvent "snapshot.discarded_unreadable" spans.Last) + let! row = snapshotRow "event_snapshot" sid + Assert.Contains("\"Total\": 10", row.Value.state) + } + + [] + member _.``state types on the same stream keep separate snapshots``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + let! _ = load (policy 1) sid + let! _ = load { policy 1 with StateType = "TallyShadow" } sid + use conn = new NpgsqlConnection(db.ConnectionString) + let! n = conn.ExecuteScalarAsync("SELECT count(*) FROM event_snapshot WHERE stream_id = @streamId", {| streamId = sid |}) + Assert.Equal(2L, n) + } + + [] + member _.``a failing snapshot write never fails the load or the append``() = task { + let broken = EventStore(db.ConnectionString, "event", jsonOpts, "broken_snapshot") + let sid = newStream () + do! appendRaw sid (added 3) + use spans = new LoadSpans() + let! (state, version) = broken.LoadState(fold, Tally.Zero, policy 1, sid) + Assert.Equal(6, state.Total) + Assert.Equal(3, version) + Assert.True(hasEvent "snapshot.write_failed" spans.Last) + Assert.Equal(ActivityStatusCode.Error, spans.Last.Status) + + let! (_, after) = broken.Transact(fold, Tally.Zero, policy 1, (fun _ -> [ Added {| amount = 4 |} ]), sid) + Assert.Equal(10, after.Total) + let! persisted = replay sid + Assert.Equal(10, persisted.Total) + } + + [] + member _.``concurrent snapshot transacts on one stream lose no events``() = task { + let sid = newStream () + let p = policy 2 + let! _ = Task.WhenAll [ for _ in 1 .. 12 -> increment p sid 1 :> Task ] + let! (state, version) = load p sid + Assert.Equal(12, state.Total) + Assert.Equal(12, version) + let! expected = replay sid + Assert.Equal(expected, state) + } + + [] + member _.``request scope serves repeated loads from memory and reflects its own appends``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + use _scope = store.BeginRequestScope() + use spans = new LoadSpans() + let! _ = load (policy 5) sid + let! _ = load (policy 5) sid + Assert.Equal([ "db"; "cache" ], spans.All |> List.map (tag "source")) + + let! _ = increment (policy 5) sid 10 + let! (state, version) = load (policy 5) sid + Assert.Equal(13, state.Total) + Assert.Equal(3, version) + Assert.Equal("cache", tag "source" spans.Last) + } + + [] + member _.``an append through the raw API invalidates the scoped snapshot state``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + use _scope = store.BeginRequestScope() + let! _ = load (policy 5) sid + let! _ = store.Transact(fold, Tally.Zero, (fun _ -> [ Added {| amount = 10 |} ]), sid) + use spans = new LoadSpans() + let! (state, version) = load (policy 5) sid + Assert.Equal(13, state.Total) + Assert.Equal(3, version) + Assert.Equal("db", tag "source" spans.Last) + } + + [] + member _.``a snapshot append invalidates the scoped raw stream``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + use _scope = store.BeginRequestScope() + let! _ = store.GetRawEventsForStream sid + let! _ = increment (policy 5) sid 10 + let! (raws, version) = store.GetRawEventsForStream sid + Assert.Equal(3, version) + Assert.Equal(3, raws.Length) + } + + [] + member _.``snapshot policy requires a snapshot table``() = + let plain = EventStore(db.ConnectionString, "event", jsonOpts) + Assert.ThrowsAsync(fun () -> + plain.LoadState(fold, Tally.Zero, policy 1, newStream ()) :> Task) + + [] + member _.``snapshot policy rejects a non-positive SnapshotEvery``() = + Assert.ThrowsAsync(fun () -> load (policy 0) (newStream ()) :> Task) diff --git a/tests/BotInfra.Tests/PostgresFixture.fs b/tests/BotInfra.Tests/PostgresFixture.fs new file mode 100644 index 0000000..db7c84e --- /dev/null +++ b/tests/BotInfra.Tests/PostgresFixture.fs @@ -0,0 +1,52 @@ +module BotInfra.Tests.PostgresFixture + +open System.Threading.Tasks +open Dapper +open Npgsql +open Testcontainers.PostgreSql +open Xunit + +/// Event + snapshot tables exactly as EVENTSTORE.md prescribes. `broken_snapshot` rejects every +/// write, to prove a failing snapshot write never fails a load or an append. +let private schemaSql = + """ +CREATE TABLE event ( + id BIGSERIAL PRIMARY KEY, + stream_id TEXT NOT NULL, + stream_version INT NOT NULL, + event_type TEXT GENERATED ALWAYS AS (data->>'Case') STORED, + data JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (stream_id, stream_version) +); + +CREATE TABLE event_snapshot ( + stream_id TEXT NOT NULL, + state_type TEXT NOT NULL, + schema_version INT NOT NULL, + stream_version INT NOT NULL, + state JSONB NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (stream_id, state_type) +); + +CREATE TABLE broken_snapshot (LIKE event_snapshot INCLUDING ALL); +ALTER TABLE broken_snapshot ADD CONSTRAINT always_fails CHECK (false); +""" + +type PostgresFixture() = + let container = PostgreSqlBuilder("postgres:17.10").Build() + + member _.ConnectionString = container.GetConnectionString() + + interface IAsyncLifetime with + member this.InitializeAsync() = + ValueTask(task { + do! container.StartAsync() + use conn = new NpgsqlConnection(this.ConnectionString) + let! _ = conn.ExecuteAsync schemaSql + return () + }) + + interface System.IAsyncDisposable with + member _.DisposeAsync() = container.DisposeAsync() diff --git a/tests/BotInfra.Tests/Program.fs b/tests/BotInfra.Tests/Program.fs new file mode 100644 index 0000000..fcc1c44 --- /dev/null +++ b/tests/BotInfra.Tests/Program.fs @@ -0,0 +1,9 @@ +namespace BotInfra.Tests + +open Xunit +open BotInfra.Tests.PostgresFixture + +// The span listener in EventStoreSnapshotTests is process-global — keep the suites sequential. +[] +[)>] +do () diff --git a/tests/BotInfra.Tests/SnapshotShapeTests.fs b/tests/BotInfra.Tests/SnapshotShapeTests.fs new file mode 100644 index 0000000..e3a0999 --- /dev/null +++ b/tests/BotInfra.Tests/SnapshotShapeTests.fs @@ -0,0 +1,41 @@ +module BotInfra.Tests.SnapshotShapeTests + +open System +open BotInfra +open Xunit + +type Kind = + | Plain + | Tagged of {| tag: string; weight: int option |} + +type Node = + { Label: string + Children: Node list } + +type Sample = + { Id: int64 + Kind: Kind + Seen: (Kind * DateTime) option + Tree: Node } + +[] +let ``describe renders nested records, unions, anonymous records, options and tuples`` () = + let expected = + "Sample{Id: System.Int64; " + + "Kind: Kind[Plain | Tagged of (Item: {tag: System.String; weight: option})]; " + + "Seen: option<(Kind[Plain | Tagged of (Item: {tag: System.String; weight: option})] * System.DateTime)>; " + + "Tree: Node{Label: System.String; Children: list<@Node>}}" + Assert.Equal(expected, SnapshotShape.describe typeof) + +type SampleWithExtraField = + { Id: int64 + Kind: Kind + Seen: (Kind * DateTime) option + Tree: Node + Extra: int option } + +[] +let ``describe changes when an optional field is added`` () = + let a = SnapshotShape.describe typeof + let b = SnapshotShape.describe typeof + Assert.NotEqual(a.Replace("SampleWithExtraField", "Sample"), b.Replace("SampleWithExtraField", "Sample")) diff --git a/tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs b/tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs new file mode 100644 index 0000000..1949cff --- /dev/null +++ b/tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs @@ -0,0 +1,88 @@ +module VahterBanBot.Unit.Tests.UserSnapshotTests + +open System +open System.Text.Json +open BotInfra +open VahterBanBot.Types +open Xunit + +let private t0 = DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc) +let private userId = 42L + +/// Exercises every UserEvent case, every Actor case and the legacy `bannedBy`-only ban shape. +let private canonicalEvents : UserEvent list = + let ban (actor: Actor option) (bannedBy: BannedBy option) (at: DateTime) = + UserBanned {| userId = userId; bannedBy = bannedBy; actor = actor; chatId = Some -100L + messageId = Some 7L; messageText = Some "spam"; bannedAt = at |} + [ UsernameChanged {| userId = userId; username = Some "alice" |} + UserReactionRecorded {| userId = userId; chatId = Some -100L; messageId = Some 5L; emoji = Some "🔥"; delta = 2 |} + UserReactionRecorded {| userId = userId; chatId = None; messageId = None; emoji = None; delta = -1 |} + ban None (Some (BannedByVahter {| vahterId = 9L; vahterUsername = Some "v"; chatId = -100L; messageId = 7; messageText = None |})) t0 + UserUnbanned {| userId = userId; unbannedBy = Some 9L; actor = None |} + ban None (Some (BannedByAI {| chatId = -100L; messageId = 8; messageText = None; modelName = "m"; promptHash = "h" |})) + (t0.AddDays 1.0) + UserUnbanned {| userId = userId; unbannedBy = None; actor = Some Actor.ML |} + ReactionTriageNotSpamSet {| userId = userId; until = t0.AddDays 2.0; actor = Actor.LLM {| modelName = "m"; promptHash = "h" |} |} + SpamProtectionGranted {| userId = userId; until = t0.AddDays 3.0; chatId = -100L; messageId = 11L; vahterId = 9L |} + SpamProtectionConsumed {| userId = userId; chatId = -100L; messageId = 12L |} + SpamProtectionRevoked {| userId = userId; reason = "budget" |} + SpamProtectionGranted {| userId = userId; until = t0.AddDays 4.0; chatId = -100L; messageId = 13L; vahterId = 9L |} + SpamProtectionConsumed {| userId = userId; chatId = -100L; messageId = 14L |} + ban (Some (Actor.Bot (Some {| botUserId = 1L; botUsername = "vahter" |}))) None (t0.AddDays 5.0) + UsernameChanged {| userId = userId; username = None |} + ban (Some (Actor.User {| userId = 9L; username = Some "v" |})) None (t0.AddDays 6.0) ] + +let private foldAll (seed: User) (events: UserEvent list) = events |> List.fold (fun s e -> User.Fold(s, e)) seed + +let private roundTrip (u: User) = + JsonSerializer.Deserialize(JsonSerializer.Serialize(u, eventJsonOpts), eventJsonOpts) + +// Append a new (SchemaVersion, shape) pair whenever User changes; never edit an existing pair. +let private pinnedShapes = + [ 1, "User{Id: System.Int64; Banned: option<(Actor[User of (Item: {userId: System.Int64; username: option}) | Bot of (Item: option<{botUserId: System.Int64; botUsername: System.String}>) | ML | LLM of (Item: {modelName: System.String; promptHash: System.String})] * System.DateTime)>; Username: option; ReactionCount: System.Int32; NotSpamUntil: option; SpamProtectionUntil: option; SpamProtectionHits: System.Int32}" ] + +// Append a new (SchemaVersion, folded state) pair whenever Fold's behaviour changes. +let private pinnedFolds = + [ 1, { Id = userId + Banned = Some (Actor.User {| userId = 9L; username = Some "v" |}, t0.AddDays 6.0) + Username = None + ReactionCount = 1 + NotSpamUntil = Some (t0.AddDays 2.0) + SpamProtectionUntil = Some (t0.AddDays 4.0) + SpamProtectionHits = 1 } ] + +[] +let ``User shape is pinned to the current snapshot SchemaVersion`` () = + let version, shape = List.last pinnedShapes + let actual = SnapshotShape.describe typeof + Assert.True( + (shape = actual && version = User.SnapshotPolicy.SchemaVersion), + $"User's shape no longer matches pin v{version} (policy is v{User.SnapshotPolicy.SchemaVersion}). " + + "Bump User.SnapshotPolicy.SchemaVersion and append a pin with the new shape:\n" + actual) + +[] +let ``User.Fold behaviour is pinned to the current snapshot SchemaVersion`` () = + let version, expected = List.last pinnedFolds + let actual = foldAll User.Zero canonicalEvents + Assert.True( + (expected = actual && version = User.SnapshotPolicy.SchemaVersion), + $"User.Fold no longer produces pin v{version} (policy is v{User.SnapshotPolicy.SchemaVersion}). " + + $"Bump User.SnapshotPolicy.SchemaVersion and append a pin with the new state:\n%A{actual}") + +[] +let ``pins use strictly increasing schema versions`` () = + for pins in [ List.map fst pinnedShapes; List.map fst pinnedFolds ] do + Assert.Equal(List.sort (List.distinct pins), pins) + +[] +let ``every intermediate User state survives the snapshot JSON round-trip`` () = + for k in 0 .. canonicalEvents.Length do + let state = foldAll User.Zero (List.take k canonicalEvents) + Assert.Equal(state, roundTrip state) + +[] +let ``snapshot plus tail equals full replay at every split point`` () = + let full = foldAll User.Zero canonicalEvents + for k in 0 .. canonicalEvents.Length do + let snapshot = roundTrip (foldAll User.Zero (List.take k canonicalEvents)) + Assert.Equal(full, foldAll snapshot (List.skip k canonicalEvents)) diff --git a/tests/VahterBanBot.Unit.Tests/VahterBanBot.Unit.Tests.fsproj b/tests/VahterBanBot.Unit.Tests/VahterBanBot.Unit.Tests.fsproj index 7dcca25..63af76f 100644 --- a/tests/VahterBanBot.Unit.Tests/VahterBanBot.Unit.Tests.fsproj +++ b/tests/VahterBanBot.Unit.Tests/VahterBanBot.Unit.Tests.fsproj @@ -24,6 +24,7 @@ + From fb291e56defa3c827f15a1d7d401980df6c6d7f2 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:09:01 +0100 Subject: [PATCH 02/10] fix(botinfra): make the request-scoped event cache safe for parallel appends; add concurrency tests A handler can append to several streams in parallel inside one request scope (vahter expires callbacks via Task.WhenAll), which raced the plain Dictionary (InvalidOperationException / corrupted state). Switch both cache maps to ConcurrentDictionary. New tests: snapshot-vs-log invariant under concurrent loads+appends, raw and snapshot writers racing on one stream, stale scoped state -> conflict -> re-read -> append, parallel appends to 200 streams inside one scope. Co-Authored-By: Claude Opus 5.5 --- src/BotInfra/EventStore.fs | 14 ++-- .../BotInfra.Tests/EventStoreSnapshotTests.fs | 83 +++++++++++++++++++ 2 files changed, 91 insertions(+), 6 deletions(-) diff --git a/src/BotInfra/EventStore.fs b/src/BotInfra/EventStore.fs index cf8f355..eb27c0b 100644 --- a/src/BotInfra/EventStore.fs +++ b/src/BotInfra/EventStore.fs @@ -1,6 +1,7 @@ namespace BotInfra open System +open System.Collections.Concurrent open System.Collections.Generic open System.Diagnostics open System.Diagnostics.Metrics @@ -87,10 +88,11 @@ type internal LoadedState = SnapshotVersion: int } /// Request-scoped identity map: raw event lists and snapshot-loaded states, keyed by stream id. +/// Concurrent: one handler may append to several streams in parallel within its scope. [] type internal RequestCache() = - member val Raws = Dictionary() - member val States = Dictionary() + member val Raws = ConcurrentDictionary() + member val States = ConcurrentDictionary() /// Append-only event store wrapper. One instance per (connection-string, event-table) /// pair. Each bot owns its own event table — this wrapper does not attempt to merge them. @@ -215,7 +217,7 @@ WHERE stream_id = @streamId AND state_type = @stateType let cacheEvict (streamId: string) = let c = scopedCache.Value if not (isNull c) then - if c.Raws.Remove streamId then EventStoreTelemetry.recordMutation "evicted" streamId + if c.Raws.TryRemove(streamId) |> fst then EventStoreTelemetry.recordMutation "evicted" streamId // Snapshot-loaded states live in the same scope; every committed append drops both views. let stateCacheTryGet (streamId: string) (stateType: string) : LoadedState option = @@ -233,15 +235,15 @@ WHERE stream_id = @streamId AND state_type = @stateType let stateCacheEvict (streamId: string) = let c = scopedCache.Value if not (isNull c) then - if c.States.Remove streamId then EventStoreTelemetry.recordMutation "evicted" streamId + if c.States.TryRemove(streamId) |> fst then EventStoreTelemetry.recordMutation "evicted" streamId /// Drops every cached view of a stream after a committed append; the caller re-populates /// whichever view it maintains, so no other view can serve pre-append data. let invalidateStream (streamId: string) = let c = scopedCache.Value if not (isNull c) then - %c.Raws.Remove streamId - %c.States.Remove streamId + %c.Raws.TryRemove streamId + %c.States.TryRemove streamId /// Reflects an append into the cache (if a scope is active) by synthesizing the new rows in /// memory, so a subsequent load in the same handle is free and reflects our own write. diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index 95559c1..9bb679e 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -3,6 +3,7 @@ module BotInfra.Tests.EventStoreSnapshotTests open System open System.Collections.Concurrent open System.Diagnostics +open System.Text.Json open System.Text.Json.Serialization open System.Threading.Tasks open Dapper @@ -92,6 +93,20 @@ type EventStoreSnapshotTests(db: PostgresFixture) = let increment p streamId amount = store.Transact(fold, Tally.Zero, p, (fun _ -> [ Added {| amount = amount |} ]), streamId) + /// The core invariant: a stored snapshot always equals the fold of the log up to its version. + let assertSnapshotMatchesLog (streamId: string) = task { + match! snapshotRow "event_snapshot" streamId with + | None -> () + | Some row -> + let! (raws, _) = EventStore(db.ConnectionString, "event", jsonOpts).GetRawEventsForStream streamId + let expected = + raws + |> List.filter (fun r -> r.stream_version <= row.stream_version) + |> List.map (fun r -> JsonSerializer.Deserialize(r.data, jsonOpts)) + |> List.fold fold Tally.Zero + Assert.Equal(expected, JsonSerializer.Deserialize(row.state, jsonOpts)) + } + [] member _.``first load replays the stream and snapshots it once the tail reaches SnapshotEvery``() = task { let sid = newStream () @@ -292,6 +307,74 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.Equal(3, raws.Length) } + [] + member _.``concurrent loads and appends never store a snapshot that disagrees with the log``() = task { + let sid = newStream () + for round in 1 .. 5 do + let work = + [ for i in 1 .. 16 -> + let p = policy (1 + i % 3) + if i % 4 = 0 then load p sid :> Task + else increment p sid 1 :> Task ] + do! Task.WhenAll work + do! assertSnapshotMatchesLog sid + let! (state, version) = load (policy 1) sid + let! expected = replay sid + Assert.Equal(expected, state) + Assert.Equal(round * 12, version) + Assert.Equal(round * 12, state.Total) + } + + [] + member _.``raw and snapshot writers racing on one stream stay consistent``() = task { + let sid = newStream () + let work = + [ for i in 1 .. 30 -> + if i % 2 = 0 then increment (policy 2) sid 1 :> Task + else store.Transact(fold, Tally.Zero, (fun _ -> [ Added {| amount = 1 |} ]), sid) :> Task ] + do! Task.WhenAll work + do! assertSnapshotMatchesLog sid + let! (state, version) = load (policy 2) sid + Assert.Equal(30, state.Total) + Assert.Equal(30, version) + } + + [] + member _.``a stale scoped state conflicts, re-reads the log and appends on top``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + let otherPod = EventStore(db.ConnectionString, "event", jsonOpts, "event_snapshot") + use _scope = store.BeginRequestScope() + let! _ = load (policy 1) sid + let! _ = otherPod.Transact(fold, Tally.Zero, policy 1, (fun _ -> [ Added {| amount = 100 |} ]), sid) + + let mutable decisions = 0 + let! (_, state) = + store.Transact(fold, Tally.Zero, policy 1, (fun _ -> decisions <- decisions + 1; [ Added {| amount = 10 |} ]), sid) + Assert.Equal(2, decisions) + Assert.Equal(113, state.Total) + let! (reloaded, version) = load (policy 1) sid + Assert.Equal(113, reloaded.Total) + Assert.Equal(4, version) + do! assertSnapshotMatchesLog sid + } + + [] + member _.``parallel appends to many streams inside one request scope are all cached correctly``() = task { + let sids = [ for _ in 1 .. 200 -> newStream () ] + use _scope = store.BeginRequestScope() + let! _ = + Task.WhenAll [ + for sid in sids do + yield task { let! _ = increment (policy 1) sid 5 in () } + yield task { let! _ = store.Transact(fold, Tally.Zero, (fun _ -> [ Added {| amount = 1 |} ]), $"{sid}:raw") in () } ] + for sid in sids do + let! (state, _) = load (policy 1) sid + Assert.Equal(5, state.Total) + let! raw = store.FoldEvents(fold, Tally.Zero, $"{sid}:raw") + Assert.Equal(1, raw.Total) + } + [] member _.``snapshot policy requires a snapshot table``() = let plain = EventStore(db.ConnectionString, "event", jsonOpts) From de17d4c4ce5725c1bb1b2abab8ecb303f04fabba Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:19:24 +0100 Subject: [PATCH 03/10] fix(botinfra): stale in-scope loads can no longer overwrite fresher cached state A load that read the DB before a parallel append committed could put its pre-append result back into the request cache after the append invalidated it (and leave the raw and snapshot views disagreeing). The request cache now serializes access and keeps a per-stream generation: invalidation bumps it and a put only lands if the generation is unchanged since the caller's read. An appender re-populates under the generation its own commit produced. Disposing a scope also disables its cache, so fire-and-forget work that captured it falls through to the DB instead of serving request-era state. Co-Authored-By: Claude Opus 5.5 --- src/BotInfra/EventStore.fs | 210 ++++++++++-------- .../BotInfra.Tests/EventStoreSnapshotTests.fs | 69 ++++++ 2 files changed, 190 insertions(+), 89 deletions(-) diff --git a/src/BotInfra/EventStore.fs b/src/BotInfra/EventStore.fs index eb27c0b..541d8cc 100644 --- a/src/BotInfra/EventStore.fs +++ b/src/BotInfra/EventStore.fs @@ -1,7 +1,6 @@ namespace BotInfra open System -open System.Collections.Concurrent open System.Collections.Generic open System.Diagnostics open System.Diagnostics.Metrics @@ -87,12 +86,45 @@ type internal LoadedState = Version: int SnapshotVersion: int } -/// Request-scoped identity map: raw event lists and snapshot-loaded states, keyed by stream id. -/// Concurrent: one handler may append to several streams in parallel within its scope. +/// Request-scoped identity map (raw streams + snapshot-loaded states). Access is serialized; a put +/// only lands if its stream wasn't invalidated since the caller observed `Generation`. [] type internal RequestCache() = - member val Raws = ConcurrentDictionary() - member val States = ConcurrentDictionary() + let gate = obj () + let raws = Dictionary() + let states = Dictionary() + let generations = Dictionary() + let mutable disposed = false + let generation streamId = + match generations.TryGetValue streamId with + | true, g -> g + | _ -> 0L + let tryFind (d: Dictionary) streamId = + match d.TryGetValue streamId with + | true, v when not disposed -> Some v + | _ -> None + + member _.Generation(streamId: string) = lock gate (fun () -> generation streamId) + member _.TryGetRaws(streamId: string) = lock gate (fun () -> tryFind raws streamId) + member _.TryGetState(streamId: string) = lock gate (fun () -> tryFind states streamId) + member _.PutRaws(streamId: string, observed: int64, entry) = + lock gate (fun () -> if not disposed && generation streamId = observed then raws[streamId] <- entry) + member _.PutState(streamId: string, observed: int64, entry) = + lock gate (fun () -> if not disposed && generation streamId = observed then states[streamId] <- entry) + /// Drops both views of the stream; returns the new generation and whether anything was cached. + member _.Invalidate(streamId: string) : int64 * bool = + lock gate (fun () -> + let next = generation streamId + 1L + generations[streamId] <- next + let removedRaws = raws.Remove streamId + let removedState = states.Remove streamId + next, (removedRaws || removedState)) + /// Work that outlives the scope (fire-and-forget) keeps the reference; it must stop caching. + member _.Dispose() = + lock gate (fun () -> + disposed <- true + raws.Clear() + states.Clear()) /// Append-only event store wrapper. One instance per (connection-string, event-table) /// pair. Each bot owns its own event table — this wrapper does not attempt to merge them. @@ -202,55 +234,50 @@ WHERE stream_id = @streamId AND state_type = @stateType // synthesized in-memory below without a re-read. let scopedCache = AsyncLocal() - let cacheTryGet (streamId: string) : (RawEvent list * int) voption = - let c = scopedCache.Value - if isNull c then ValueNone - else - match c.Raws.TryGetValue streamId with - | true, v -> ValueSome v - | _ -> ValueNone + /// The current scope's generation for `streamId`, captured before a DB read (None outside a scope). + let observeGeneration (streamId: string) : int64 option = + match scopedCache.Value with + | null -> None + | c -> Some (c.Generation streamId) - let cachePut (streamId: string) (entry: RawEvent list * int) = - let c = scopedCache.Value - if not (isNull c) then c.Raws[streamId] <- entry + let cacheTryGet (streamId: string) : (RawEvent list * int) option = + match scopedCache.Value with + | null -> None + | c -> c.TryGetRaws streamId - let cacheEvict (streamId: string) = - let c = scopedCache.Value - if not (isNull c) then - if c.Raws.TryRemove(streamId) |> fst then EventStoreTelemetry.recordMutation "evicted" streamId + let cachePut (streamId: string) (observed: int64 option) (entry: RawEvent list * int) = + match scopedCache.Value, observed with + | null, _ | _, None -> () + | c, Some g -> c.PutRaws(streamId, g, entry) // Snapshot-loaded states live in the same scope; every committed append drops both views. let stateCacheTryGet (streamId: string) (stateType: string) : LoadedState option = - let c = scopedCache.Value - if isNull c then None - else - match c.States.TryGetValue streamId with - | true, e when e.StateType = stateType -> Some e - | _ -> None - - let stateCachePut (streamId: string) (entry: LoadedState) = - let c = scopedCache.Value - if not (isNull c) then c.States[streamId] <- entry - - let stateCacheEvict (streamId: string) = - let c = scopedCache.Value - if not (isNull c) then - if c.States.TryRemove(streamId) |> fst then EventStoreTelemetry.recordMutation "evicted" streamId - - /// Drops every cached view of a stream after a committed append; the caller re-populates - /// whichever view it maintains, so no other view can serve pre-append data. - let invalidateStream (streamId: string) = - let c = scopedCache.Value - if not (isNull c) then - %c.Raws.TryRemove streamId - %c.States.TryRemove streamId + match scopedCache.Value with + | null -> None + | c -> c.TryGetState streamId |> Option.filter (fun e -> e.StateType = stateType) + + let stateCachePut (streamId: string) (observed: int64 option) (entry: LoadedState) = + match scopedCache.Value, observed with + | null, _ | _, None -> () + | c, Some g -> c.PutState(streamId, g, entry) + + /// Drops every cached view of a stream; returns the generation the caller may put under. + let invalidateStream (streamId: string) : int64 option = + match scopedCache.Value with + | null -> None + | c -> Some (fst (c.Invalidate streamId)) + + /// Same as `invalidateStream` after a lost version race, counted as an eviction. + let cacheEvict (streamId: string) = + match scopedCache.Value with + | null -> () + | c -> if snd (c.Invalidate streamId) then EventStoreTelemetry.recordMutation "evicted" streamId /// Reflects an append into the cache (if a scope is active) by synthesizing the new rows in /// memory, so a subsequent load in the same handle is free and reflects our own write. /// Synthesized rows carry meaningful `data` + `stream_version` only (see note above). - let cacheAppend (streamId: string) (priorRaws: RawEvent list) (baseVersion: int) (newEvents: 'TEvent list) = - let c = scopedCache.Value - if not (isNull c) then + let cacheAppend (streamId: string) (observed: int64 option) (priorRaws: RawEvent list) (baseVersion: int) (newEvents: 'TEvent list) = + if observed.IsSome then let synthesized = newEvents |> List.mapi (fun i e -> @@ -261,7 +288,7 @@ WHERE stream_id = @streamId AND state_type = @stateType data = JsonSerializer.Serialize<'TEvent>(e, jsonOptions) created_at = Unchecked.defaultof }) let newVersion = baseVersion + List.length newEvents - c.Raws[streamId] <- (priorRaws @ synthesized, newVersion) + cachePut streamId observed (priorRaws @ synthesized, newVersion) EventStoreTelemetry.recordMutation "appended" streamId /// Upserts a snapshot. A failure is recorded (metric + span error) and swallowed: snapshots @@ -297,6 +324,7 @@ WHERE stream_id = @streamId AND state_type = @stateType return entry | None -> use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.load") + let observed = observeGeneration streamId use conn = new NpgsqlConnection(connString) let keyArgs = {| streamId = streamId; stateType = policy.StateType; schemaVersion = policy.SchemaVersion |} @@ -357,19 +385,20 @@ WHERE stream_id = @streamId AND state_type = @stateType let entry = { StateType = policy.StateType; State = box state Version = version; SnapshotVersion = snapshotVersion } - stateCachePut streamId entry + stateCachePut streamId observed entry return entry } let readStream (streamId: string) : Task = task { match cacheTryGet streamId with - | ValueSome (events, version) -> + | Some (events, version) -> use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.load") EventStoreTelemetry.recordLoad activity "cache" streamId version (List.length events) return events, version - | ValueNone -> + | None -> use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.load") + let observed = observeGeneration streamId use conn = new NpgsqlConnection(connString) let! rows = conn.QueryAsync(selectAllSql, {| streamId = streamId |}) let events = List.ofSeq rows @@ -378,7 +407,7 @@ WHERE stream_id = @streamId AND state_type = @stateType |> List.tryLast |> Option.map (fun e -> e.stream_version) |> Option.defaultValue 0 - cachePut streamId (events, version) + cachePut streamId observed (events, version) EventStoreTelemetry.recordLoad activity "db" streamId version (List.length events) return events, version } @@ -399,13 +428,37 @@ WHERE stream_id = @streamId AND state_type = @stateType return insertedCount } + /// Inserts + optional in-TX projection; on commit invalidates the scoped cache and returns the + /// generation the appender may re-populate it under. + let tryAppendCore (streamId: string) (expectedVersion: int) (events: 'TEvent list) + (projection: NpgsqlConnection -> NpgsqlTransaction -> Task) : Task> = + task { + if events.IsEmpty then return Ok (observeGeneration streamId) + else + use conn = new NpgsqlConnection(connString) + do! conn.OpenAsync() + use! tx = conn.BeginTransactionAsync() + let! inserted = insertEvents conn tx streamId expectedVersion events + if inserted < events.Length then + do! tx.RollbackAsync() + return Error ConcurrencyConflict + else + do! projection conn tx + do! tx.CommitAsync() + return Ok (invalidateStream streamId) + } + /// Begins a request-scoped identity-map scope. Repeated loads of the same stream within the /// returned scope are served from memory; dispose restores the prior scope (supports nesting). /// Intended to wrap one update handle — see the webhook entry point. member _.BeginRequestScope() : IDisposable = let prev = scopedCache.Value - scopedCache.Value <- RequestCache() - { new IDisposable with member _.Dispose() = scopedCache.Value <- prev } + let cache = RequestCache() + scopedCache.Value <- cache + { new IDisposable with + member _.Dispose() = + cache.Dispose() + scopedCache.Value <- prev } /// Returns the highest stream_version for the given stream, or 0 if the stream is empty. member _.GetStreamVersion(streamId: string) : Task = @@ -489,19 +542,9 @@ ORDER BY stream_id, stream_version (streamId: string, expectedVersion: int, events: 'TEvent list) : Task> = task { - if events.IsEmpty then return Ok () - else - use conn = new NpgsqlConnection(connString) - do! conn.OpenAsync() - use! tx = conn.BeginTransactionAsync() - let! inserted = insertEvents conn tx streamId expectedVersion events - if inserted < events.Length then - do! tx.RollbackAsync() - return Error ConcurrencyConflict - else - do! tx.CommitAsync() - invalidateStream streamId - return Ok () + match! tryAppendCore streamId expectedVersion events (fun _ _ -> Task.CompletedTask) with + | Ok _ -> return Ok () + | Error e -> return Error e } /// Same as TryAppend, but runs `projection conn tx` after the inserts succeed and @@ -514,20 +557,9 @@ ORDER BY stream_id, stream_version projection: NpgsqlConnection -> NpgsqlTransaction -> Task) : Task> = task { - if events.IsEmpty then return Ok () - else - use conn = new NpgsqlConnection(connString) - do! conn.OpenAsync() - use! tx = conn.BeginTransactionAsync() - let! inserted = insertEvents conn tx streamId expectedVersion events - if inserted < events.Length then - do! tx.RollbackAsync() - return Error ConcurrencyConflict - else - do! projection conn tx - do! tx.CommitAsync() - invalidateStream streamId - return Ok () + match! tryAppendCore streamId expectedVersion events projection with + | Ok _ -> return Ok () + | Error e -> return Error e } /// Read-decide-append-retry loop with optimistic concurrency. On conflict the @@ -551,9 +583,9 @@ ORDER BY stream_id, stream_version if newEvents.IsEmpty then result <- ValueSome ([], state) else - match! this.TryAppend(streamId, version, newEvents) with - | Ok _ -> - cacheAppend streamId raws version newEvents + match! tryAppendCore streamId version newEvents (fun _ _ -> Task.CompletedTask) with + | Ok observed -> + cacheAppend streamId observed raws version newEvents let finalState = newEvents |> List.fold fold state result <- ValueSome (newEvents, finalState) | Error ConcurrencyConflict -> @@ -589,9 +621,9 @@ ORDER BY stream_id, stream_version match projection with | Some p -> p | None -> fun _ _ -> Task.CompletedTask - match! this.TryAppendWithProjection(streamId, version, newEvents, proj) with - | Ok _ -> - cacheAppend streamId raws version newEvents + match! tryAppendCore streamId version newEvents proj with + | Ok observed -> + cacheAppend streamId observed raws version newEvents let finalState = newEvents |> List.fold fold state result <- ValueSome (newEvents, finalState) | Error ConcurrencyConflict -> @@ -631,8 +663,8 @@ ORDER BY stream_id, stream_version match projection with | Some p -> p | None -> fun _ _ -> Task.CompletedTask - match! this.TryAppendWithProjection(streamId, loaded.Version, newEvents, proj) with - | Ok _ -> + match! tryAppendCore streamId loaded.Version newEvents proj with + | Ok observed -> let finalState = newEvents |> List.fold fold state let newVersion = loaded.Version + newEvents.Length let! snapshotVersion = @@ -643,12 +675,12 @@ ORDER BY stream_id, stream_version return if written then newVersion else loaded.SnapshotVersion } else Task.FromResult loaded.SnapshotVersion - stateCachePut streamId + stateCachePut streamId observed { loaded with State = box finalState; Version = newVersion; SnapshotVersion = snapshotVersion } result <- Some (newEvents, finalState) | Error ConcurrencyConflict -> // Stale read lost the version race — the retry must re-read from the DB. - stateCacheEvict streamId + cacheEvict streamId return result.Value } diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index 9bb679e..60133da 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -5,6 +5,7 @@ open System.Collections.Concurrent open System.Diagnostics open System.Text.Json open System.Text.Json.Serialization +open System.Threading open System.Threading.Tasks open Dapper open Npgsql @@ -56,6 +57,19 @@ type private LoadSpans() = let private tag (key: string) (a: Activity) = a.GetTagItem key |> string let private hasEvent (name: string) (a: Activity) = a.Events |> Seq.exists (fun e -> e.Name = name) +/// A fold that parks its first call until released — holds a load between its query and its cache put. +let private parkedFold () = + let entered = new SemaphoreSlim(0) + let release = new SemaphoreSlim(0) + let mutable first = true + let f s e = + if first then + first <- false + %entered.Release() + release.Wait() + fold s e + f, entered, release + type SnapshotRow = { schema_version: int; stream_version: int; state: string } type EventStoreSnapshotTests(db: PostgresFixture) = @@ -375,6 +389,61 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.Equal(1, raw.Total) } + [] + member _.``a load that read before a parallel append cannot overwrite the fresher cached state``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + use _scope = store.BeginRequestScope() + let parked, entered, release = parkedFold () + let slowLoad = store.LoadState(parked, Tally.Zero, policy 100, sid) + do! entered.WaitAsync() + let! _ = increment (policy 100) sid 10 + %release.Release() + let! (_, slowVersion) = slowLoad + Assert.Equal(2, slowVersion) + let! (state, version) = load (policy 100) sid + Assert.Equal(13, state.Total) + Assert.Equal(3, version) + } + + [] + member _.``a load that read before a parallel raw append cannot leave the two cached views disagreeing``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + use _scope = store.BeginRequestScope() + let parked, entered, release = parkedFold () + let slowLoad = store.LoadState(parked, Tally.Zero, policy 100, sid) + do! entered.WaitAsync() + let! _ = store.Transact(fold, Tally.Zero, (fun _ -> [ Added {| amount = 10 |} ]), sid) + %release.Release() + let! _ = slowLoad + let! (state, version) = load (policy 100) sid + let! (_, rawVersion) = store.GetRawEventsForStream sid + Assert.Equal(3, rawVersion) + Assert.Equal(3, version) + Assert.Equal(13, state.Total) + } + + [] + member _.``work outliving its request scope stops using the scope's cache``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + let otherPod = EventStore(db.ConnectionString, "event", jsonOpts, "event_snapshot") + let go = new SemaphoreSlim(0) + let scope = store.BeginRequestScope() + let! _ = load (policy 100) sid + let background = task { + do! go.WaitAsync() + return! load (policy 100) sid + } + scope.Dispose() + let! _ = otherPod.Transact(fold, Tally.Zero, (fun _ -> [ Added {| amount = 10 |} ]), sid) + %go.Release() + let! (state, version) = background + Assert.Equal(3, version) + Assert.Equal(13, state.Total) + } + [] member _.``snapshot policy requires a snapshot table``() = let plain = EventStore(db.ConnectionString, "event", jsonOpts) From 9911b1348ce1a8ec52e57251a626a14128ad195b Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:20:22 +0100 Subject: [PATCH 04/10] fix(botinfra): report append-path snapshot write failures on their own span The post-commit snapshot write ran under the caller's span, so a deliberately swallowed failure marked the business span as Error. It now runs under a dedicated eventStore.snapshotWrite span. Co-Authored-By: Claude Opus 5.5 --- src/BotInfra/EventStore.fs | 3 ++- .../BotInfra.Tests/EventStoreSnapshotTests.fs | 21 +++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/src/BotInfra/EventStore.fs b/src/BotInfra/EventStore.fs index 541d8cc..898df19 100644 --- a/src/BotInfra/EventStore.fs +++ b/src/BotInfra/EventStore.fs @@ -670,8 +670,9 @@ ORDER BY stream_id, stream_version let! snapshotVersion = if newVersion - loaded.SnapshotVersion >= policy.SnapshotEvery then task { + use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.snapshotWrite") use conn = new NpgsqlConnection(connString) - let! written = tryWriteSnapshot conn Activity.Current policy streamId newVersion finalState + let! written = tryWriteSnapshot conn activity policy streamId newVersion finalState return if written then newVersion else loaded.SnapshotVersion } else Task.FromResult loaded.SnapshotVersion diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index 60133da..dba720b 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -266,6 +266,27 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.Equal(10, persisted.Total) } + [] + member _.``a failed snapshot write on append is reported on its own span, never on the caller's``() = task { + use callerSource = new ActivitySource("BotInfra.Tests.Caller") + let stopped = ConcurrentQueue() + use listener = + new ActivityListener( + ShouldListenTo = (fun s -> s.Name = callerSource.Name || s.Name = "BotInfra.EventStore"), + Sample = (fun _ -> ActivitySamplingResult.AllDataAndRecorded), + ActivityStopped = (fun a -> stopped.Enqueue a)) + ActivitySource.AddActivityListener listener + let broken = EventStore(db.ConnectionString, "event", jsonOpts, "broken_snapshot") + let caller = callerSource.StartActivity("caller") + let! _ = broken.Transact(fold, Tally.Zero, policy 1, (fun _ -> [ Added {| amount = 1 |} ]), newStream ()) + caller.Stop() + Assert.Equal(ActivityStatusCode.Unset, caller.Status) + Assert.Empty(caller.Events) + let write = stopped |> Seq.find (fun a -> a.OperationName = "eventStore.snapshotWrite") + Assert.Equal(ActivityStatusCode.Error, write.Status) + Assert.True(hasEvent "snapshot.write_failed" write) + } + [] member _.``concurrent snapshot transacts on one stream lose no events``() = task { let sid = newStream () From 6ef66aedbe10f8d25f6440c8fe2e5c1fecf8bb39 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:22:01 +0100 Subject: [PATCH 05/10] fix(botinfra): a snapshot discard can neither fail the load nor get stuck Deleting a known-bad snapshot (ahead of its log / unreadable) could throw and fail the load, and if it didn't run the monotonic upsert guard kept the bad row forever. The discard path now force-overwrites the row with the replayed state; only an empty stream deletes, and that delete is recorded + swallowed. Co-Authored-By: Claude Opus 5.5 --- src/BotInfra/EventStore.fs | 62 +++++++++++++++---- .../BotInfra.Tests/EventStoreSnapshotTests.fs | 32 ++++++++++ tests/BotInfra.Tests/PostgresFixture.fs | 9 ++- 3 files changed, 88 insertions(+), 15 deletions(-) diff --git a/src/BotInfra/EventStore.fs b/src/BotInfra/EventStore.fs index 898df19..b3ebcf1 100644 --- a/src/BotInfra/EventStore.fs +++ b/src/BotInfra/EventStore.fs @@ -29,7 +29,7 @@ module internal EventStoreTelemetry = let private snapshotOpsCounter = meter.CreateCounter( "eventstore_snapshot_ops_total", "ops", - "Aggregate-snapshot outcomes, tagged by state_type and op (hit|miss|written|discarded_ahead|discarded_unreadable|write_failed)") + "Aggregate-snapshot outcomes, tagged by state_type and op (hit|miss|written|discarded_ahead|discarded_unreadable|write_failed|delete_failed)") /// Records a load and tags the (possibly null) load span with its provenance. let recordLoad (activity: Activity) (source: string) (streamId: string) (version: int) (eventCount: int) = @@ -217,6 +217,19 @@ ON CONFLICT (stream_id, state_type) DO UPDATE OR {snapshotTable}.stream_version < EXCLUDED.stream_version """ + // Discard path only: the stored row is known-bad (ahead of its log or unreadable), so the guard + // that keeps snapshots monotonic must not protect it. + let snapshotForceWriteSql = + $""" +INSERT INTO {snapshotTable} (stream_id, state_type, schema_version, stream_version, state) +VALUES (@streamId, @stateType, @schemaVersion, @streamVersion, @state::JSONB) +ON CONFLICT (stream_id, state_type) DO UPDATE + SET schema_version = EXCLUDED.schema_version, + stream_version = EXCLUDED.stream_version, + state = EXCLUDED.state, + updated_at = now() +""" + let snapshotDeleteSql = $""" DELETE FROM {snapshotTable} @@ -293,13 +306,13 @@ WHERE stream_id = @streamId AND state_type = @stateType /// Upserts a snapshot. A failure is recorded (metric + span error) and swallowed: snapshots /// are a cache, so a failed write must never fail the load or the committed append. - let tryWriteSnapshot (conn: NpgsqlConnection) (activity: Activity) (policy: SnapshotPolicy) + let tryWriteSnapshot (conn: NpgsqlConnection) (activity: Activity) (policy: SnapshotPolicy) (force: bool) (streamId: string) (version: int) (state: 'State) : Task = task { try let json = JsonSerializer.Serialize<'State>(state, jsonOptions) let! _ = - conn.ExecuteAsync(snapshotWriteSql, + conn.ExecuteAsync((if force then snapshotForceWriteSql else snapshotWriteSql), {| streamId = streamId; stateType = policy.StateType; schemaVersion = policy.SchemaVersion streamVersion = version; state = json |}) EventStoreTelemetry.recordSnapshot activity policy.StateType "written" @@ -311,8 +324,24 @@ WHERE stream_id = @streamId AND state_type = @stateType return false } + /// Removes a known-bad snapshot of an empty stream; failures are recorded and swallowed. + let tryDeleteSnapshot (conn: NpgsqlConnection) (activity: Activity) (policy: SnapshotPolicy) + (streamId: string) (badVersion: int) : Task = + task { + try + let! _ = + conn.ExecuteAsync(snapshotDeleteSql, + {| streamId = streamId; stateType = policy.StateType + schemaVersion = policy.SchemaVersion; streamVersion = badVersion |}) + () + with ex -> + EventStoreTelemetry.recordSnapshot activity policy.StateType "delete_failed" + if not (isNull activity) then + %activity.AddException(ex).SetStatus(ActivityStatusCode.Error, "snapshot delete failed") + } + /// Loads aggregate state as snapshot + tail. A snapshot that is ahead of its log or can't be - /// deserialized is deleted and the stream is replayed from scratch. + /// deserialized is ignored, the stream replayed from scratch, and the row overwritten. let loadSnapshotted (fold: 'State -> 'TEvent -> 'State) (zero: 'State) (policy: SnapshotPolicy) (streamId: string) : Task = task { @@ -357,12 +386,8 @@ WHERE stream_id = @streamId AND state_type = @stateType | Ok None -> EventStoreTelemetry.recordSnapshot activity policy.StateType "miss" return zero, 0, tail - | Error (op, badVersion) -> + | Error (op, _) -> EventStoreTelemetry.recordSnapshot activity policy.StateType op - let! _ = - conn.ExecuteAsync(snapshotDeleteSql, - {| streamId = streamId; stateType = policy.StateType - schemaVersion = policy.SchemaVersion; streamVersion = badVersion |}) let! raws = conn.QueryAsync(selectAllSql, {| streamId = streamId |}) return zero, 0, (raws |> Seq.map (fun r -> r.stream_version, r.data) |> List.ofSeq) } @@ -374,12 +399,23 @@ WHERE stream_id = @streamId AND state_type = @stateType | Some (v, _) -> v | None -> baseVersion let! snapshotVersion = - if version - baseVersion >= policy.SnapshotEvery then + match fromSnapshot with + | Error (_, badVersion) when version = 0 -> + task { + do! tryDeleteSnapshot conn activity policy streamId badVersion + return 0 + } + | Error _ -> + task { + let! written = tryWriteSnapshot conn activity policy true streamId version state + return if written then version else 0 + } + | Ok _ when version - baseVersion >= policy.SnapshotEvery -> task { - let! written = tryWriteSnapshot conn activity policy streamId version state + let! written = tryWriteSnapshot conn activity policy false streamId version state return if written then version else baseVersion } - else Task.FromResult baseVersion + | Ok _ -> Task.FromResult baseVersion EventStoreTelemetry.recordLoad activity "db" streamId version (List.length events) if not (isNull activity) then %activity.SetTag("snapshot_version", baseVersion) let entry = @@ -672,7 +708,7 @@ ORDER BY stream_id, stream_version task { use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.snapshotWrite") use conn = new NpgsqlConnection(connString) - let! written = tryWriteSnapshot conn activity policy streamId newVersion finalState + let! written = tryWriteSnapshot conn activity policy false streamId newVersion finalState return if written then newVersion else loaded.SnapshotVersion } else Task.FromResult loaded.SnapshotVersion diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index dba720b..401a980 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -222,6 +222,38 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.Equal(Some 4, row |> Option.map _.stream_version) } + [] + member _.``a discarded snapshot that cannot be deleted neither fails the load nor sticks around``() = task { + let stubborn = EventStore(db.ConnectionString, "event", jsonOpts, "undeletable_snapshot") + let sid = newStream () + do! appendRaw sid (added 4) + let! _ = stubborn.LoadState(fold, Tally.Zero, policy 1, sid) + do! exec "UPDATE undeletable_snapshot SET stream_version = 999 WHERE stream_id = @streamId" sid + + let! (state, version) = stubborn.LoadState(fold, Tally.Zero, policy 1, sid) + Assert.Equal(10, state.Total) + Assert.Equal(4, version) + let! row = snapshotRow "undeletable_snapshot" sid + Assert.Equal(Some 4, row |> Option.map _.stream_version) + } + + [] + member _.``a snapshot of a stream with no events is dropped, even when deleting it fails``() = task { + for table in [ "event_snapshot"; "undeletable_snapshot" ] do + let s = EventStore(db.ConnectionString, "event", jsonOpts, table) + let sid = newStream () + use conn = new NpgsqlConnection(db.ConnectionString) + let! _ = + conn.ExecuteAsync( + $"INSERT INTO {table} (stream_id, state_type, schema_version, stream_version, state) VALUES (@streamId, 'Tally', 1, 3, '{{\"Total\": 9, \"Count\": 3}}')", + {| streamId = sid |}) + let! (state, version) = s.LoadState(fold, Tally.Zero, policy 1, sid) + Assert.Equal(Tally.Zero, state) + Assert.Equal(0, version) + let! row = snapshotRow table sid + Assert.Equal((if table = "event_snapshot" then None else Some 3), row |> Option.map _.stream_version) + } + [] member _.``an unreadable snapshot is discarded and rebuilt``() = task { let sid = newStream () diff --git a/tests/BotInfra.Tests/PostgresFixture.fs b/tests/BotInfra.Tests/PostgresFixture.fs index db7c84e..6596f9b 100644 --- a/tests/BotInfra.Tests/PostgresFixture.fs +++ b/tests/BotInfra.Tests/PostgresFixture.fs @@ -6,8 +6,8 @@ open Npgsql open Testcontainers.PostgreSql open Xunit -/// Event + snapshot tables exactly as EVENTSTORE.md prescribes. `broken_snapshot` rejects every -/// write, to prove a failing snapshot write never fails a load or an append. +/// Event + snapshot tables exactly as EVENTSTORE.md prescribes, plus two sabotaged snapshot tables: +/// `broken_snapshot` rejects every write, `undeletable_snapshot` rejects every delete. let private schemaSql = """ CREATE TABLE event ( @@ -32,6 +32,11 @@ CREATE TABLE event_snapshot ( CREATE TABLE broken_snapshot (LIKE event_snapshot INCLUDING ALL); ALTER TABLE broken_snapshot ADD CONSTRAINT always_fails CHECK (false); + +CREATE TABLE undeletable_snapshot (LIKE event_snapshot INCLUDING ALL); +CREATE FUNCTION refuse_delete() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN RAISE EXCEPTION 'deletes refused'; END $$; +CREATE TRIGGER refuse_delete BEFORE DELETE ON undeletable_snapshot FOR EACH ROW EXECUTE FUNCTION refuse_delete(); """ type PostgresFixture() = From 47bceeb16eba664035ca4f51cd00a77fb8b439bb Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:24:02 +0100 Subject: [PATCH 06/10] test(vahter): pin User.Fold transcript + snapshot JSON, not just the final state The old pin only compared the final state of one sequence, so a Fold change whose effect is overwritten later (e.g. UserUnbanned clearing NotSpamUntil) passed. The pin is now a transcript of every intermediate state plus each event applied to Zero and to a fully-populated state, serialized with the snapshot JSON options; the sequence gains no-grant/not-banned/negative edge cases. Catches that mutation, a ReactionCount clamp, and a JSON-options change. Co-Authored-By: Claude Opus 5.5 --- .../UserSnapshotTests.fs | 112 +++++++++++++++--- 1 file changed, 98 insertions(+), 14 deletions(-) diff --git a/tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs b/tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs index 1949cff..1134275 100644 --- a/tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs +++ b/tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs @@ -2,6 +2,7 @@ module VahterBanBot.Unit.Tests.UserSnapshotTests open System open System.Text.Json +open Microsoft.FSharp.Reflection open BotInfra open VahterBanBot.Types open Xunit @@ -9,12 +10,17 @@ open Xunit let private t0 = DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc) let private userId = 42L -/// Exercises every UserEvent case, every Actor case and the legacy `bannedBy`-only ban shape. +/// Every UserEvent case, every Actor case, the legacy `bannedBy`-only ban shape, and edge cases +/// (consume/revoke without a grant, unban while not banned, negative reaction count). let private canonicalEvents : UserEvent list = let ban (actor: Actor option) (bannedBy: BannedBy option) (at: DateTime) = UserBanned {| userId = userId; bannedBy = bannedBy; actor = actor; chatId = Some -100L messageId = Some 7L; messageText = Some "spam"; bannedAt = at |} - [ UsernameChanged {| userId = userId; username = Some "alice" |} + [ SpamProtectionConsumed {| userId = userId; chatId = -100L; messageId = 1L |} + SpamProtectionRevoked {| userId = userId; reason = "killed" |} + UserUnbanned {| userId = userId; unbannedBy = Some 9L; actor = None |} + UserReactionRecorded {| userId = userId; chatId = None; messageId = None; emoji = None; delta = -3 |} + UsernameChanged {| userId = userId; username = Some "alice" |} UserReactionRecorded {| userId = userId; chatId = Some -100L; messageId = Some 5L; emoji = Some "🔥"; delta = 2 |} UserReactionRecorded {| userId = userId; chatId = None; messageId = None; emoji = None; delta = -1 |} ban None (Some (BannedByVahter {| vahterId = 9L; vahterUsername = Some "v"; chatId = -100L; messageId = 7; messageText = None |})) t0 @@ -41,15 +47,93 @@ let private roundTrip (u: User) = let private pinnedShapes = [ 1, "User{Id: System.Int64; Banned: option<(Actor[User of (Item: {userId: System.Int64; username: option}) | Bot of (Item: option<{botUserId: System.Int64; botUsername: System.String}>) | ML | LLM of (Item: {modelName: System.String; promptHash: System.String})] * System.DateTime)>; Username: option; ReactionCount: System.Int32; NotSpamUntil: option; SpamProtectionUntil: option; SpamProtectionHits: System.Int32}" ] -// Append a new (SchemaVersion, folded state) pair whenever Fold's behaviour changes. +/// Fully-populated state, so a Fold case that touches a field it shouldn't shows up in the transcript. +let private richSeed = + { Id = 7L + Banned = Some (Actor.LLM {| modelName = "seed"; promptHash = "seed" |}, t0.AddDays -1.0) + Username = Some "seed" + ReactionCount = 7 + NotSpamUntil = Some (t0.AddDays 9.0) + SpamProtectionUntil = Some (t0.AddDays 9.0) + SpamProtectionHits = 2 } + +/// Every intermediate state of the canonical sequence plus each event applied to Zero and to a +/// fully-populated state, as snapshot JSON — pins Fold's behaviour AND the stored format. +let private foldTranscript () = + let json (u: User) = JsonSerializer.Serialize(u, eventJsonOpts) + let case (e: UserEvent) = (FSharpValue.GetUnionFields(e, typeof) |> fst).Name + let fold s e = User.Fold(s, e) + [ for e, s in List.zip canonicalEvents (List.tail (List.scan fold User.Zero canonicalEvents)) do + yield $"seq {case e} -> {json s}" + for e in canonicalEvents do + yield $"zero {case e} -> {json (fold User.Zero e)}" + yield $"rich {case e} -> {json (fold richSeed e)}" ] + |> String.concat "\n" + +// Append a new (SchemaVersion, transcript) pair whenever Fold or the snapshot JSON changes. let private pinnedFolds = - [ 1, { Id = userId - Banned = Some (Actor.User {| userId = 9L; username = Some "v" |}, t0.AddDays 6.0) - Username = None - ReactionCount = 1 - NotSpamUntil = Some (t0.AddDays 2.0) - SpamProtectionUntil = Some (t0.AddDays 4.0) - SpamProtectionHits = 1 } ] + [ 1, + String.concat "\n" [ + """seq SpamProtectionConsumed -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":1}""" + """seq SpamProtectionRevoked -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """seq UserUnbanned -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """seq UserReactionRecorded -> {"Id":42,"ReactionCount":-3,"SpamProtectionHits":0}""" + """seq UsernameChanged -> {"Id":42,"Username":"alice","ReactionCount":-3,"SpamProtectionHits":0}""" + """seq UserReactionRecorded -> {"Id":42,"Username":"alice","ReactionCount":-1,"SpamProtectionHits":0}""" + """seq UserReactionRecorded -> {"Id":42,"Username":"alice","ReactionCount":-2,"SpamProtectionHits":0}""" + """seq UserBanned -> {"Id":42,"Banned":[{"Case":"User","userId":9,"username":"v"},"2026-01-01T00:00:00Z"],"Username":"alice","ReactionCount":-2,"SpamProtectionHits":0}""" + """seq UserUnbanned -> {"Id":42,"Username":"alice","ReactionCount":-2,"SpamProtectionHits":0}""" + """seq UserBanned -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"m","promptHash":"h"},"2026-01-02T00:00:00Z"],"Username":"alice","ReactionCount":-2,"SpamProtectionHits":0}""" + """seq UserUnbanned -> {"Id":42,"Username":"alice","ReactionCount":-2,"SpamProtectionHits":0}""" + """seq ReactionTriageNotSpamSet -> {"Id":42,"Username":"alice","ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionHits":0}""" + """seq SpamProtectionGranted -> {"Id":42,"Username":"alice","ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-04T00:00:00Z","SpamProtectionHits":0}""" + """seq SpamProtectionConsumed -> {"Id":42,"Username":"alice","ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-04T00:00:00Z","SpamProtectionHits":1}""" + """seq SpamProtectionRevoked -> {"Id":42,"Username":"alice","ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionHits":0}""" + """seq SpamProtectionGranted -> {"Id":42,"Username":"alice","ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-05T00:00:00Z","SpamProtectionHits":0}""" + """seq SpamProtectionConsumed -> {"Id":42,"Username":"alice","ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-05T00:00:00Z","SpamProtectionHits":1}""" + """seq UserBanned -> {"Id":42,"Banned":[{"Case":"Bot","Item":{"botUserId":1,"botUsername":"vahter"}},"2026-01-06T00:00:00Z"],"Username":"alice","ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-05T00:00:00Z","SpamProtectionHits":1}""" + """seq UsernameChanged -> {"Id":42,"Banned":[{"Case":"Bot","Item":{"botUserId":1,"botUsername":"vahter"}},"2026-01-06T00:00:00Z"],"ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-05T00:00:00Z","SpamProtectionHits":1}""" + """seq UserBanned -> {"Id":42,"Banned":[{"Case":"User","userId":9,"username":"v"},"2026-01-07T00:00:00Z"],"ReactionCount":-2,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-05T00:00:00Z","SpamProtectionHits":1}""" + """zero SpamProtectionConsumed -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":1}""" + """rich SpamProtectionConsumed -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":3}""" + """zero SpamProtectionRevoked -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """rich SpamProtectionRevoked -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":0}""" + """zero UserUnbanned -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UserUnbanned -> {"Id":42,"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserReactionRecorded -> {"Id":42,"ReactionCount":-3,"SpamProtectionHits":0}""" + """rich UserReactionRecorded -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":4,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UsernameChanged -> {"Id":42,"Username":"alice","ReactionCount":0,"SpamProtectionHits":0}""" + """rich UsernameChanged -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"alice","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserReactionRecorded -> {"Id":42,"ReactionCount":2,"SpamProtectionHits":0}""" + """rich UserReactionRecorded -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":9,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserReactionRecorded -> {"Id":42,"ReactionCount":-1,"SpamProtectionHits":0}""" + """rich UserReactionRecorded -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":6,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserBanned -> {"Id":42,"Banned":[{"Case":"User","userId":9,"username":"v"},"2026-01-01T00:00:00Z"],"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UserBanned -> {"Id":42,"Banned":[{"Case":"User","userId":9,"username":"v"},"2026-01-01T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserUnbanned -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UserUnbanned -> {"Id":42,"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserBanned -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"m","promptHash":"h"},"2026-01-02T00:00:00Z"],"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UserBanned -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"m","promptHash":"h"},"2026-01-02T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserUnbanned -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UserUnbanned -> {"Id":42,"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero ReactionTriageNotSpamSet -> {"Id":42,"ReactionCount":0,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionHits":0}""" + """rich ReactionTriageNotSpamSet -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-03T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero SpamProtectionGranted -> {"Id":42,"ReactionCount":0,"SpamProtectionUntil":"2026-01-04T00:00:00Z","SpamProtectionHits":0}""" + """rich SpamProtectionGranted -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-04T00:00:00Z","SpamProtectionHits":0}""" + """zero SpamProtectionConsumed -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":1}""" + """rich SpamProtectionConsumed -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":3}""" + """zero SpamProtectionRevoked -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """rich SpamProtectionRevoked -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":0}""" + """zero SpamProtectionGranted -> {"Id":42,"ReactionCount":0,"SpamProtectionUntil":"2026-01-05T00:00:00Z","SpamProtectionHits":0}""" + """rich SpamProtectionGranted -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-05T00:00:00Z","SpamProtectionHits":0}""" + """zero SpamProtectionConsumed -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":1}""" + """rich SpamProtectionConsumed -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":3}""" + """zero UserBanned -> {"Id":42,"Banned":[{"Case":"Bot","Item":{"botUserId":1,"botUsername":"vahter"}},"2026-01-06T00:00:00Z"],"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UserBanned -> {"Id":42,"Banned":[{"Case":"Bot","Item":{"botUserId":1,"botUsername":"vahter"}},"2026-01-06T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UsernameChanged -> {"Id":42,"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UsernameChanged -> {"Id":42,"Banned":[{"Case":"LLM","modelName":"seed","promptHash":"seed"},"2025-12-31T00:00:00Z"],"ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" + """zero UserBanned -> {"Id":42,"Banned":[{"Case":"User","userId":9,"username":"v"},"2026-01-07T00:00:00Z"],"ReactionCount":0,"SpamProtectionHits":0}""" + """rich UserBanned -> {"Id":42,"Banned":[{"Case":"User","userId":9,"username":"v"},"2026-01-07T00:00:00Z"],"Username":"seed","ReactionCount":7,"NotSpamUntil":"2026-01-10T00:00:00Z","SpamProtectionUntil":"2026-01-10T00:00:00Z","SpamProtectionHits":2}""" ] ] [] let ``User shape is pinned to the current snapshot SchemaVersion`` () = @@ -61,13 +145,13 @@ let ``User shape is pinned to the current snapshot SchemaVersion`` () = + "Bump User.SnapshotPolicy.SchemaVersion and append a pin with the new shape:\n" + actual) [] -let ``User.Fold behaviour is pinned to the current snapshot SchemaVersion`` () = +let ``User.Fold behaviour and snapshot JSON are pinned to the current snapshot SchemaVersion`` () = let version, expected = List.last pinnedFolds - let actual = foldAll User.Zero canonicalEvents + let actual = foldTranscript () Assert.True( (expected = actual && version = User.SnapshotPolicy.SchemaVersion), - $"User.Fold no longer produces pin v{version} (policy is v{User.SnapshotPolicy.SchemaVersion}). " - + $"Bump User.SnapshotPolicy.SchemaVersion and append a pin with the new state:\n%A{actual}") + $"User.Fold / snapshot JSON no longer match pin v{version} (policy is v{User.SnapshotPolicy.SchemaVersion}). " + + "Bump User.SnapshotPolicy.SchemaVersion and append a pin with the new transcript:\n" + actual) [] let ``pins use strictly increasing schema versions`` () = From 0e6a0d094987758db2dc4de53b30f0ebdcbcee04 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:24:58 +0100 Subject: [PATCH 07/10] fix(botinfra): SnapshotShape renders generic types structurally MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Generic BCL/FSharp.Core types fell back to FullName, which embeds assembly-qualified type arguments with versions — a runtime upgrade would fail shape pins with no real change. Render them as Name recursively. Co-Authored-By: Claude Opus 5.5 --- src/BotInfra/EventSnapshot.fs | 4 ++++ tests/BotInfra.Tests/SnapshotShapeTests.fs | 12 ++++++++++++ 2 files changed, 16 insertions(+) diff --git a/src/BotInfra/EventSnapshot.fs b/src/BotInfra/EventSnapshot.fs index 7765b38..c27a2a3 100644 --- a/src/BotInfra/EventSnapshot.fs +++ b/src/BotInfra/EventSnapshot.fs @@ -51,6 +51,10 @@ module SnapshotShape = |> sprintf "[%s]" %visiting.Remove t name + body + elif t.IsGenericType then + let name = t.Name.Substring(0, t.Name.IndexOf '`') + let args = t.GetGenericArguments() |> Array.map (render visiting) |> String.concat ", " + $"{name}<{args}>" elif t.IsEnum then let names = Enum.GetNames t |> String.concat "," $"{t.Name}<{names}>" diff --git a/tests/BotInfra.Tests/SnapshotShapeTests.fs b/tests/BotInfra.Tests/SnapshotShapeTests.fs index e3a0999..085fdd0 100644 --- a/tests/BotInfra.Tests/SnapshotShapeTests.fs +++ b/tests/BotInfra.Tests/SnapshotShapeTests.fs @@ -39,3 +39,15 @@ let ``describe changes when an optional field is added`` () = let a = SnapshotShape.describe typeof let b = SnapshotShape.describe typeof Assert.NotEqual(a.Replace("SampleWithExtraField", "Sample"), b.Replace("SampleWithExtraField", "Sample")) + +type WithCollections = + { Tags: Map + Seen: Set + Pair: System.Collections.Generic.KeyValuePair } + +[] +let ``describe renders generic collections without assembly versions`` () = + Assert.Equal( + "WithCollections{Tags: FSharpMap>; Seen: FSharpSet; " + + "Pair: KeyValuePair}", + SnapshotShape.describe typeof) From 5fea641660940d3370a2b7ea81e10dd3e405b964 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:26:42 +0100 Subject: [PATCH 08/10] fix(botinfra): count guard-blocked snapshot writes as skipped, not written Co-Authored-By: Claude Opus 5.5 --- src/BotInfra/EventStore.fs | 10 ++++---- .../BotInfra.Tests/EventStoreSnapshotTests.fs | 23 +++++++++++++++++++ 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/src/BotInfra/EventStore.fs b/src/BotInfra/EventStore.fs index b3ebcf1..dded465 100644 --- a/src/BotInfra/EventStore.fs +++ b/src/BotInfra/EventStore.fs @@ -29,7 +29,7 @@ module internal EventStoreTelemetry = let private snapshotOpsCounter = meter.CreateCounter( "eventstore_snapshot_ops_total", "ops", - "Aggregate-snapshot outcomes, tagged by state_type and op (hit|miss|written|discarded_ahead|discarded_unreadable|write_failed|delete_failed)") + "Aggregate-snapshot outcomes, tagged by state_type and op (hit|miss|written|skipped|discarded_ahead|discarded_unreadable|write_failed|delete_failed)") /// Records a load and tags the (possibly null) load span with its provenance. let recordLoad (activity: Activity) (source: string) (streamId: string) (version: int) (eventCount: int) = @@ -311,12 +311,14 @@ WHERE stream_id = @streamId AND state_type = @stateType task { try let json = JsonSerializer.Serialize<'State>(state, jsonOptions) - let! _ = + let! affected = conn.ExecuteAsync((if force then snapshotForceWriteSql else snapshotWriteSql), {| streamId = streamId; stateType = policy.StateType; schemaVersion = policy.SchemaVersion streamVersion = version; state = json |}) - EventStoreTelemetry.recordSnapshot activity policy.StateType "written" - return true + // 0 rows = the monotonic guard kept a newer snapshot another writer stored first. + let op = if affected > 0 then "written" else "skipped" + EventStoreTelemetry.recordSnapshot activity policy.StateType op + return affected > 0 with ex -> EventStoreTelemetry.recordSnapshot activity policy.StateType "write_failed" if not (isNull activity) then diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index 401a980..1a7391b 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -497,6 +497,29 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.Equal(13, state.Total) } + [] + member _.``a slow writer never regresses a newer snapshot and does not count as written``() = task { + let sid = newStream () + do! appendRaw sid (added 2) + let otherPod = EventStore(db.ConnectionString, "event", jsonOpts, "event_snapshot") + use spans = new LoadSpans() + let parked, entered, release = parkedFold () + let slowLoad = store.LoadState(parked, Tally.Zero, policy 1, sid) + do! entered.WaitAsync() + do! appendRaw sid (added 3) + let! _ = otherPod.LoadState(fold, Tally.Zero, policy 1, sid) + %release.Release() + let! (_, slowVersion) = slowLoad + Assert.Equal(2, slowVersion) + + let! row = snapshotRow "event_snapshot" sid + Assert.Equal(Some 5, row |> Option.map _.stream_version) + do! assertSnapshotMatchesLog sid + let slowSpan = spans.All |> List.find (fun a -> tag "stream_version" a = "2" && hasEvent "snapshot.miss" a) + Assert.True(hasEvent "snapshot.skipped" slowSpan) + Assert.False(hasEvent "snapshot.written" slowSpan) + } + [] member _.``snapshot policy requires a snapshot table``() = let plain = EventStore(db.ConnectionString, "event", jsonOpts) From 7e01f5d90cc5ffb376e9385cccfa44fea3c92db3 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:27:50 +0100 Subject: [PATCH 09/10] fix(vahter,botinfra): in-place event rewrites drop the stream's snapshots A data-fix migration that UPDATEs/DELETEs event rows left existing snapshots looking valid (same schema version, not ahead of the log) while disagreeing with the rewritten history. event_snapshot_invalidate triggers (row-level on UPDATE/DELETE, statement-level on TRUNCATE) now drop affected snapshots; the bot role never mutates events, so they only fire in migrations. Documented along with the known costs (two-schema rollout replays, TryAppend gaps). Co-Authored-By: Claude Opus 5.5 --- src/BotInfra/EVENTSTORE.md | 42 ++++++++++++++++--- .../migrations/V53__event_snapshot.sql | 15 +++++++ .../BotInfra.Tests/EventStoreSnapshotTests.fs | 16 +++++++ tests/BotInfra.Tests/PostgresFixture.fs | 15 +++++++ 4 files changed, 83 insertions(+), 5 deletions(-) diff --git a/src/BotInfra/EVENTSTORE.md b/src/BotInfra/EVENTSTORE.md index d66984a..585544d 100644 --- a/src/BotInfra/EVENTSTORE.md +++ b/src/BotInfra/EVENTSTORE.md @@ -164,6 +164,21 @@ CREATE TABLE event_snapshot ( PRIMARY KEY (stream_id, state_type) ); GRANT SELECT, INSERT, UPDATE, DELETE ON event_snapshot TO ; + +-- Snapshots are only valid for an append-only log: any in-place rewrite of events drops them. +CREATE OR REPLACE FUNCTION event_snapshot_invalidate() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'TRUNCATE' THEN + TRUNCATE event_snapshot; + ELSE + DELETE FROM event_snapshot WHERE stream_id = OLD.stream_id OR stream_id = NEW.stream_id; + END IF; + RETURN NULL; +END $$; +CREATE OR REPLACE TRIGGER event_snapshot_invalidate AFTER UPDATE OR DELETE ON event + FOR EACH ROW EXECUTE FUNCTION event_snapshot_invalidate(); +CREATE OR REPLACE TRIGGER event_snapshot_invalidate_truncate AFTER TRUNCATE ON event + FOR EACH STATEMENT EXECUTE FUNCTION event_snapshot_invalidate(); ``` ```fsharp @@ -187,8 +202,14 @@ Rules the implementation guarantees: commit so it can never roll back events. - **Stale snapshots are detected and discarded.** A row whose `schema_version` differs from the policy is ignored (and overwritten by the next write). A row - that is ahead of its log or doesn't deserialize is deleted and the stream is - replayed (`op="discarded_ahead" | "discarded_unreadable"`). + that is ahead of its log or doesn't deserialize is ignored, the stream is + replayed and the row force-overwritten (`op="discarded_ahead" | + "discarded_unreadable"`); a failure there is recorded, never thrown. +- **Rewriting history drops snapshots.** The triggers above delete a stream's + snapshots on any `UPDATE`/`DELETE` of its events (and `TRUNCATE` clears the + table), so a data-fix migration can never leave a snapshot that disagrees + with the rewritten log. The bot role never updates events, so the triggers + only fire during migrations. - **Snapshots only move forward** within a schema version (upsert guard), so a slow writer can't regress a newer snapshot. A different schema version (new code or a rollback) always overwrites. @@ -200,7 +221,15 @@ Rules the implementation guarantees: get snapshotted lazily on their first slow load, no backfill needed. - **Request scope.** `BeginRequestScope` caches snapshot-loaded states next to raw streams; every committed append (through any API) drops both cached views - of that stream, then the appending path re-populates its own. + of that stream, then the appending path re-populates its own. A load only + publishes into the cache if no append invalidated the stream since it read + the DB, so a slow parallel load can't put pre-append state back. A disposed + scope stops caching, so fire-and-forget work that captured it reads the DB. +- **Known costs.** While pods with two different `SchemaVersion`s run side by + side (rolling deploy of a bump), each treats the other's row as a miss, so + long streams replay until the rollout finishes. `TryAppend` with an + `expectedVersion` above the head can create a version gap; the Transact + loops never do — prefer them. ### Versioning — the one thing you must not forget @@ -211,8 +240,11 @@ snapshots. Enforce it with two pin tests next to the aggregate (see `tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs`): 1. `SnapshotShape.describe typeof` pinned per schema version. -2. The fold of a canonical event sequence (covering every case) pinned per - schema version. +2. A fold *transcript* pinned per schema version: the snapshot JSON of every + intermediate state of a canonical sequence (every event case plus edge + cases), and of each event applied to `Zero` and to a fully-populated state. + Pinning only a final state misses changes whose effect is overwritten later; + pinning JSON also catches serializer-option changes. Both assert the *latest* pin's version equals the policy's, so changing the type or the fold fails until the version is bumped and a new pin appended. diff --git a/src/vahter-bot/migrations/V53__event_snapshot.sql b/src/vahter-bot/migrations/V53__event_snapshot.sql index 36a9276..864c94b 100644 --- a/src/vahter-bot/migrations/V53__event_snapshot.sql +++ b/src/vahter-bot/migrations/V53__event_snapshot.sql @@ -10,4 +10,19 @@ CREATE TABLE IF NOT EXISTS event_snapshot ( PRIMARY KEY (stream_id, state_type) ); +-- Snapshots are only valid for an append-only log: any in-place rewrite of events drops them. +CREATE OR REPLACE FUNCTION event_snapshot_invalidate() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'TRUNCATE' THEN + TRUNCATE event_snapshot; + ELSE + DELETE FROM event_snapshot WHERE stream_id = OLD.stream_id OR stream_id = NEW.stream_id; + END IF; + RETURN NULL; +END $$; +CREATE OR REPLACE TRIGGER event_snapshot_invalidate AFTER UPDATE OR DELETE ON event + FOR EACH ROW EXECUTE FUNCTION event_snapshot_invalidate(); +CREATE OR REPLACE TRIGGER event_snapshot_invalidate_truncate AFTER TRUNCATE ON event + FOR EACH STATEMENT EXECUTE FUNCTION event_snapshot_invalidate(); + GRANT SELECT, INSERT, UPDATE, DELETE ON event_snapshot TO vahter_bot_ban_service; diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index 1a7391b..09082a8 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -520,6 +520,22 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.False(hasEvent "snapshot.written" slowSpan) } + [] + member _.``rewriting or deleting events in place drops the stream's snapshots``() = task { + let sid = newStream () + do! appendRaw sid (added 4) + let! _ = load (policy 1) sid + do! exec """UPDATE event SET data = jsonb_set(data, '{amount}', '100') WHERE stream_id = @streamId AND stream_version = 1""" sid + let! (rewritten, _) = load (policy 100) sid + Assert.Equal(109, rewritten.Total) + + let! _ = load (policy 1) sid + do! exec "DELETE FROM event WHERE stream_id = @streamId AND stream_version = 4" sid + let! (trimmed, version) = load (policy 100) sid + Assert.Equal(105, trimmed.Total) + Assert.Equal(3, version) + } + [] member _.``snapshot policy requires a snapshot table``() = let plain = EventStore(db.ConnectionString, "event", jsonOpts) diff --git a/tests/BotInfra.Tests/PostgresFixture.fs b/tests/BotInfra.Tests/PostgresFixture.fs index 6596f9b..29dab1b 100644 --- a/tests/BotInfra.Tests/PostgresFixture.fs +++ b/tests/BotInfra.Tests/PostgresFixture.fs @@ -30,6 +30,21 @@ CREATE TABLE event_snapshot ( PRIMARY KEY (stream_id, state_type) ); +-- Snapshots are only valid for an append-only log: any in-place rewrite of events drops them. +CREATE OR REPLACE FUNCTION event_snapshot_invalidate() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'TRUNCATE' THEN + TRUNCATE event_snapshot; + ELSE + DELETE FROM event_snapshot WHERE stream_id = OLD.stream_id OR stream_id = NEW.stream_id; + END IF; + RETURN NULL; +END $$; +CREATE OR REPLACE TRIGGER event_snapshot_invalidate AFTER UPDATE OR DELETE ON event + FOR EACH ROW EXECUTE FUNCTION event_snapshot_invalidate(); +CREATE OR REPLACE TRIGGER event_snapshot_invalidate_truncate AFTER TRUNCATE ON event + FOR EACH STATEMENT EXECUTE FUNCTION event_snapshot_invalidate(); + CREATE TABLE broken_snapshot (LIKE event_snapshot INCLUDING ALL); ALTER TABLE broken_snapshot ADD CONSTRAINT always_fails CHECK (false); From bdaf0410c5e77c9c2d1ee9f0ed5b0cb66fce1d09 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Fri, 25 Sep 2026 11:28:29 +0100 Subject: [PATCH 10/10] test(botinfra): two schema versions racing on one stream stay consistent Co-Authored-By: Claude Opus 5.5 --- .../BotInfra.Tests/EventStoreSnapshotTests.fs | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index 09082a8..71fafd3 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -536,6 +536,29 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.Equal(3, version) } + [] + member _.``pods on two schema versions racing on one stream each read only their own snapshots``() = task { + let sid = newStream () + let v1 = policy 2 + let v2 = { policy 2 with SchemaVersion = 2 } + let newPod = EventStore(db.ConnectionString, "event", jsonOpts, "event_snapshot") + let work = + [ for i in 1 .. 24 -> + match i % 4 with + | 0 -> load v1 sid :> Task + | 1 -> increment v1 sid 1 :> Task + | 2 -> newPod.LoadState(fold, Tally.Zero, v2, sid) :> Task + | _ -> newPod.Transact(fold, Tally.Zero, v2, (fun _ -> [ Added {| amount = 1 |} ]), sid) :> Task ] + do! Task.WhenAll work + do! assertSnapshotMatchesLog sid + let! expected = replay sid + let! (old, _) = load v1 sid + let! (fresh, _) = newPod.LoadState(fold, Tally.Zero, v2, sid) + Assert.Equal(expected, old) + Assert.Equal(expected, fresh) + Assert.Equal(12, expected.Total) + } + [] member _.``snapshot policy requires a snapshot table``() = let plain = EventStore(db.ConnectionString, "event", jsonOpts)