From 4de38c23c5ea3fa26661862672f3d4a782fa8778 Mon Sep 17 00:00:00 2001 From: bc19sam-afk Date: Wed, 29 Jul 2026 15:56:07 +1000 Subject: [PATCH 1/2] test(manager-server): define usage snapshot protocol contract --- .../httpapi/server_usage_snapshot_test.go | 87 +++++++++ .../repository/usageevent/snapshot_test.go | 176 ++++++++++++++++++ .../internal/service/usage/snapshot_test.go | 148 +++++++++++++++ 3 files changed, 411 insertions(+) create mode 100644 apps/manager-server/internal/httpapi/server_usage_snapshot_test.go create mode 100644 apps/manager-server/internal/repository/usageevent/snapshot_test.go create mode 100644 apps/manager-server/internal/service/usage/snapshot_test.go diff --git a/apps/manager-server/internal/httpapi/server_usage_snapshot_test.go b/apps/manager-server/internal/httpapi/server_usage_snapshot_test.go new file mode 100644 index 000000000..4899eedb2 --- /dev/null +++ b/apps/manager-server/internal/httpapi/server_usage_snapshot_test.go @@ -0,0 +1,87 @@ +package httpapi + +import ( + "context" + "net/http" + "strings" + "testing" + + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/store" + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/testutil" + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/usage" +) + +func TestServerUsageSnapshotProtocolRequiresAdminAndPreservesLegacyRoutes(t *testing.T) { + cpa := testutil.NewCPAMock(t) + setup := &store.Setup{CPAUpstreamURL: cpa.URL(), ManagementKey: "management-key", Queue: "usage", PopSide: "right"} + handler, db := newCompatHandler(t, testutil.NewConfig(t), setup) + + first := compatEvent("same-shape-a", 10) + second := compatEvent("same-shape-b", 10) + first.RequestID = "request-a" + second.RequestID = "request-b" + first.RawJSON = `{"authorization":"must-not-leak"}` + first.FailBody = "Bearer must-not-leak" + if _, err := db.InsertEvents(context.Background(), []usage.Event{first, second}); err != nil { + t.Fatalf("insert events: %v", err) + } + + unauthorized := testutil.Request(t, handler, http.MethodGet, "/v1/management/usage/events?limit=1", "", "") + testutil.RequireStatus(t, unauthorized, http.StatusUnauthorized) + cpaKey := testutil.Request(t, handler, http.MethodGet, "/v1/management/usage/events?limit=1", "", "management-key") + testutil.RequireStatus(t, cpaKey, http.StatusUnauthorized) + + firstRR := testutil.Request(t, handler, http.MethodGet, "/v1/management/usage/events?limit=1", "", testutil.AdminKey) + testutil.RequireStatus(t, firstRR, http.StatusOK) + var page struct { + ProtocolVersion int `json:"protocol_version"` + SnapshotID string `json:"snapshot_id"` + MaxEventID int64 `json:"max_event_id"` + RowCount int64 `json:"row_count"` + Digest string `json:"digest"` + DigestAlgorithm string `json:"digest_algorithm"` + Complete bool `json:"complete"` + NextCursor string `json:"next_cursor"` + Events []struct { + EventID int64 `json:"event_id"` + RequestID string `json:"request_id"` + EventHash string `json:"event_hash"` + } `json:"events"` + } + testutil.DecodeJSON(t, firstRR, &page) + if page.ProtocolVersion != 1 || page.SnapshotID == "" || page.MaxEventID != 2 || page.RowCount != 2 || + page.Digest == "" || page.DigestAlgorithm != "sha256:event-id-event-hash:v1" || page.Complete || + page.NextCursor == "" || len(page.Events) != 1 || page.Events[0].EventID == 0 || + page.Events[0].RequestID != "request-a" || page.Events[0].EventHash != "same-shape-a" { + t.Fatalf("snapshot page = %#v", page) + } + if strings.Contains(firstRR.Body.String(), "must-not-leak") { + t.Fatalf("snapshot exposed sensitive fields: %s", firstRR.Body.String()) + } + + badCursor := page.NextCursor[:len(page.NextCursor)-1] + "x" + tamperedRR := testutil.Request( + t, + handler, + http.MethodGet, + "/v1/management/usage/events?snapshot_id="+page.SnapshotID+"&cursor="+badCursor+"&limit=1", + "", + testutil.AdminKey, + ) + testutil.RequireStatus(t, tamperedRR, http.StatusBadRequest) + if !strings.Contains(tamperedRR.Body.String(), `"code":"usage_snapshot_invalid_cursor"`) { + t.Fatalf("tampered cursor body = %s", tamperedRR.Body.String()) + } + + legacyUsage := testutil.Request(t, handler, http.MethodGet, "/v0/management/usage", "", testutil.AdminKey) + testutil.RequireStatus(t, legacyUsage, http.StatusOK) + if !strings.Contains(legacyUsage.Body.String(), `"total_requests":2`) || strings.Contains(legacyUsage.Body.String(), `"event_id"`) { + t.Fatalf("legacy usage body changed incompatibly: %s", legacyUsage.Body.String()) + } + legacyExport := testutil.Request(t, handler, http.MethodGet, "/v0/management/usage/export", "", testutil.AdminKey) + testutil.RequireStatus(t, legacyExport, http.StatusOK) + if !strings.Contains(legacyExport.Body.String(), `"event_hash":"same-shape-a"`) || + strings.Contains(legacyExport.Body.String(), `"event_id"`) { + t.Fatalf("legacy export body changed incompatibly: %s", legacyExport.Body.String()) + } +} diff --git a/apps/manager-server/internal/repository/usageevent/snapshot_test.go b/apps/manager-server/internal/repository/usageevent/snapshot_test.go new file mode 100644 index 000000000..9a9cd7dce --- /dev/null +++ b/apps/manager-server/internal/repository/usageevent/snapshot_test.go @@ -0,0 +1,176 @@ +package usageevent + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "path/filepath" + "testing" + + sqliterepo "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/sqlite" + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/usage" +) + +func TestUsageSnapshotFreezesHighWaterAndUsesStableEventIDs(t *testing.T) { + db, err := sqliterepo.Open(filepath.Join(t.TempDir(), "usage.sqlite")) + if err != nil { + t.Fatalf("open database: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + repo := New(db) + + first := snapshotTestEvent("same-shape-a", 1) + second := snapshotTestEvent("same-shape-b", 1) + third := snapshotTestEvent("later", 2) + if _, err := repo.InsertBatch(context.Background(), []usage.Event{first, second, third}); err != nil { + t.Fatalf("insert initial events: %v", err) + } + + snapshot, err := repo.CaptureSnapshot(context.Background()) + if err != nil { + t.Fatalf("capture snapshot: %v", err) + } + if snapshot.MaxEventID != 3 || snapshot.RowCount != 3 { + t.Fatalf("snapshot = %#v", snapshot) + } + + if _, err := repo.InsertBatch(context.Background(), []usage.Event{snapshotTestEvent("after-high-water", 3)}); err != nil { + t.Fatalf("insert concurrent event: %v", err) + } + + var events []SnapshotEvent + var afterID int64 + for { + page, err := repo.SnapshotPage(context.Background(), snapshot.MaxEventID, afterID, 2) + if err != nil { + t.Fatalf("read page after %d: %v", afterID, err) + } + events = append(events, page.Events...) + if !page.HasMore { + break + } + afterID = page.Events[len(page.Events)-1].EventID + } + + if len(events) != 3 { + t.Fatalf("snapshot event count = %d, events = %#v", len(events), events) + } + for index, event := range events { + wantID := int64(index + 1) + if event.EventID != wantID { + t.Fatalf("event %d id = %d, want %d", index, event.EventID, wantID) + } + if event.EventHash == "after-high-water" { + t.Fatalf("snapshot included concurrent event: %#v", event) + } + } + if events[0].TimestampMS != events[1].TimestampMS || events[0].Model != events[1].Model || + events[0].Endpoint != events[1].Endpoint || events[0].EventID == events[1].EventID { + t.Fatalf("same-shape events lost distinct identity: %#v", events[:2]) + } + if got := snapshotDigest(events); got != snapshot.Digest { + t.Fatalf("digest = %q, want %q", got, snapshot.Digest) + } + + next, err := repo.CaptureSnapshot(context.Background()) + if err != nil { + t.Fatalf("capture next snapshot: %v", err) + } + if next.MaxEventID != 4 || next.RowCount != 4 || next.Digest == snapshot.Digest { + t.Fatalf("next snapshot = %#v", next) + } +} + +func TestUsageSnapshotPagesAllRowsBeyondLegacy50000Limit(t *testing.T) { + db, err := sqliterepo.Open(filepath.Join(t.TempDir(), "usage.sqlite")) + if err != nil { + t.Fatalf("open database: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + tx, err := db.BeginTx(context.Background(), nil) + if err != nil { + t.Fatalf("begin fixture transaction: %v", err) + } + stmt, err := tx.Prepare(`insert into usage_events ( + event_hash, timestamp_ms, timestamp, model, endpoint, input_tokens, output_tokens, total_tokens, created_at_ms + ) values (?, ?, '2026-01-01T00:00:00Z', 'gpt-test', 'POST /v1/responses', 1, 2, 3, ?)`) + if err != nil { + _ = tx.Rollback() + t.Fatalf("prepare fixture insert: %v", err) + } + const rowCount = 50_001 + for index := 1; index <= rowCount; index++ { + if _, err := stmt.Exec(fmt.Sprintf("snapshot-%05d", index), index, index); err != nil { + _ = stmt.Close() + _ = tx.Rollback() + t.Fatalf("insert fixture %d: %v", index, err) + } + } + if err := stmt.Close(); err != nil { + _ = tx.Rollback() + t.Fatalf("close fixture statement: %v", err) + } + if err := tx.Commit(); err != nil { + t.Fatalf("commit fixtures: %v", err) + } + + repo := New(db) + snapshot, err := repo.CaptureSnapshot(context.Background()) + if err != nil { + t.Fatalf("capture snapshot: %v", err) + } + if snapshot.RowCount != rowCount || snapshot.MaxEventID != rowCount { + t.Fatalf("snapshot = %#v", snapshot) + } + + hash := sha256.New() + var delivered int64 + var afterID int64 + for { + page, err := repo.SnapshotPage(context.Background(), snapshot.MaxEventID, afterID, 4096) + if err != nil { + t.Fatalf("read page after %d: %v", afterID, err) + } + for _, event := range page.Events { + _, _ = fmt.Fprintf(hash, "%d\x00%s\n", event.EventID, event.EventHash) + afterID = event.EventID + delivered++ + } + if !page.HasMore { + break + } + } + if delivered != rowCount || afterID != rowCount { + t.Fatalf("delivered = %d, last id = %d", delivered, afterID) + } + if got := "sha256:" + hex.EncodeToString(hash.Sum(nil)); got != snapshot.Digest { + t.Fatalf("digest = %q, want %q", got, snapshot.Digest) + } +} + +func snapshotTestEvent(hash string, timestampMS int64) usage.Event { + return usage.Event{ + EventHash: hash, + TimestampMS: timestampMS, + Timestamp: "2026-01-01T00:00:00Z", + Model: "gpt-test", + Endpoint: "POST /v1/responses", + Source: "masked-source", + InputTokens: 1, + OutputTokens: 2, + TotalTokens: 3, + FailBody: "Bearer must-not-leak", + RawJSON: `{"authorization":"must-not-leak"}`, + CreatedAtMS: timestampMS, + } +} + +func snapshotDigest(events []SnapshotEvent) string { + hash := sha256.New() + for _, event := range events { + _, _ = fmt.Fprintf(hash, "%d\x00%s\n", event.EventID, event.EventHash) + } + return "sha256:" + hex.EncodeToString(hash.Sum(nil)) +} diff --git a/apps/manager-server/internal/service/usage/snapshot_test.go b/apps/manager-server/internal/service/usage/snapshot_test.go new file mode 100644 index 000000000..73530f1e5 --- /dev/null +++ b/apps/manager-server/internal/service/usage/snapshot_test.go @@ -0,0 +1,148 @@ +package usage + +import ( + "context" + "path/filepath" + "reflect" + "testing" + "time" + + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/store" + usageparser "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/usage" +) + +func TestSnapshotProtocolReplaysAndRejectsTamperedCrossSnapshotCursor(t *testing.T) { + now := time.Unix(1_800_000_000, 0) + db, err := store.Open(filepath.Join(t.TempDir(), "usage.sqlite")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if _, err := db.InsertEvents(context.Background(), []usageparser.Event{ + snapshotServiceEvent("event-a", 1), + snapshotServiceEvent("event-b", 2), + }); err != nil { + t.Fatalf("insert events: %v", err) + } + + service := New(db, WithSnapshotProtocol(SnapshotProtocolConfig{ + SigningKey: []byte("0123456789abcdef0123456789abcdef"), + TTL: time.Hour, + Now: func() time.Time { return now }, + })) + first, err := service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{Limit: 1}) + if err != nil { + t.Fatalf("read first page: %v", err) + } + if first.Complete || first.NextCursor == "" || first.RowCount != 2 || first.MaxEventID != 2 || len(first.Events) != 1 { + t.Fatalf("first page = %#v", first) + } + if first.Events[0].EventID == 0 || first.Events[0].EventHash != "event-a" { + t.Fatalf("first event = %#v", first.Events[0]) + } + + replay, err := service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{ + SnapshotID: first.SnapshotID, + Limit: 1, + }) + if err != nil { + t.Fatalf("replay first page: %v", err) + } + if !reflect.DeepEqual(replay, first) { + t.Fatalf("replay = %#v, want %#v", replay, first) + } + + last, err := service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{ + SnapshotID: first.SnapshotID, + Cursor: first.NextCursor, + Limit: 1, + }) + if err != nil { + t.Fatalf("read final page: %v", err) + } + if !last.Complete || last.NextCursor != "" || len(last.Events) != 1 || last.Events[0].EventHash != "event-b" { + t.Fatalf("last page = %#v", last) + } + + tampered := first.NextCursor[:len(first.NextCursor)-1] + "x" + _, err = service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{ + SnapshotID: first.SnapshotID, + Cursor: tampered, + Limit: 1, + }) + requireSnapshotErrorCode(t, err, SnapshotErrorInvalidCursor) + + other, err := service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{Limit: 1}) + if err != nil { + t.Fatalf("create other snapshot: %v", err) + } + if other.SnapshotID == first.SnapshotID { + t.Fatal("independent snapshots reused the same snapshot_id") + } + _, err = service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{ + SnapshotID: other.SnapshotID, + Cursor: first.NextCursor, + Limit: 1, + }) + requireSnapshotErrorCode(t, err, SnapshotErrorInvalidCursor) +} + +func TestSnapshotProtocolExpiresSnapshotAndCursor(t *testing.T) { + now := time.Unix(1_800_000_000, 0) + db, err := store.Open(filepath.Join(t.TempDir(), "usage.sqlite")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if _, err := db.InsertEvents(context.Background(), []usageparser.Event{ + snapshotServiceEvent("event-a", 1), + snapshotServiceEvent("event-b", 2), + }); err != nil { + t.Fatalf("insert events: %v", err) + } + + service := New(db, WithSnapshotProtocol(SnapshotProtocolConfig{ + SigningKey: []byte("0123456789abcdef0123456789abcdef"), + TTL: time.Minute, + Now: func() time.Time { return now }, + })) + page, err := service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{Limit: 1}) + if err != nil { + t.Fatalf("create snapshot: %v", err) + } + now = now.Add(2 * time.Minute) + + _, err = service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{ + SnapshotID: page.SnapshotID, + Limit: 1, + }) + requireSnapshotErrorCode(t, err, SnapshotErrorExpired) + _, err = service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{ + SnapshotID: page.SnapshotID, + Cursor: page.NextCursor, + Limit: 1, + }) + requireSnapshotErrorCode(t, err, SnapshotErrorExpired) +} + +func snapshotServiceEvent(hash string, timestampMS int64) usageparser.Event { + return usageparser.Event{ + EventHash: hash, + TimestampMS: timestampMS, + Timestamp: "2026-01-01T00:00:00Z", + Model: "gpt-test", + Endpoint: "POST /v1/responses", + InputTokens: 1, + OutputTokens: 2, + TotalTokens: 3, + CreatedAtMS: timestampMS, + } +} + +func requireSnapshotErrorCode(t *testing.T, err error, want SnapshotErrorCode) { + t.Helper() + snapshotErr, ok := err.(*SnapshotError) + if !ok || snapshotErr.Code != want { + t.Fatalf("snapshot error = %#v, want code %q", err, want) + } +} From 824d0f963b8aceb8644c81644dcb041fb53b9b77 Mon Sep 17 00:00:00 2001 From: bc19sam-afk Date: Wed, 29 Jul 2026 15:58:38 +1000 Subject: [PATCH 2/2] test(manager-server): fail closed on snapshot drift --- .../httpapi/server_usage_snapshot_test.go | 2 +- .../repository/usageevent/snapshot_test.go | 27 ++++++++--- .../internal/service/usage/snapshot_test.go | 46 +++++++++++++++++++ 3 files changed, 68 insertions(+), 7 deletions(-) diff --git a/apps/manager-server/internal/httpapi/server_usage_snapshot_test.go b/apps/manager-server/internal/httpapi/server_usage_snapshot_test.go index 4899eedb2..6fe9f4d98 100644 --- a/apps/manager-server/internal/httpapi/server_usage_snapshot_test.go +++ b/apps/manager-server/internal/httpapi/server_usage_snapshot_test.go @@ -50,7 +50,7 @@ func TestServerUsageSnapshotProtocolRequiresAdminAndPreservesLegacyRoutes(t *tes } testutil.DecodeJSON(t, firstRR, &page) if page.ProtocolVersion != 1 || page.SnapshotID == "" || page.MaxEventID != 2 || page.RowCount != 2 || - page.Digest == "" || page.DigestAlgorithm != "sha256:event-id-event-hash:v1" || page.Complete || + page.Digest == "" || page.DigestAlgorithm != "sha256-chain:event-id-record-digest:v1" || page.Complete || page.NextCursor == "" || len(page.Events) != 1 || page.Events[0].EventID == 0 || page.Events[0].RequestID != "request-a" || page.Events[0].EventHash != "same-shape-a" { t.Fatalf("snapshot page = %#v", page) diff --git a/apps/manager-server/internal/repository/usageevent/snapshot_test.go b/apps/manager-server/internal/repository/usageevent/snapshot_test.go index 9a9cd7dce..202b4a045 100644 --- a/apps/manager-server/internal/repository/usageevent/snapshot_test.go +++ b/apps/manager-server/internal/repository/usageevent/snapshot_test.go @@ -64,6 +64,9 @@ func TestUsageSnapshotFreezesHighWaterAndUsesStableEventIDs(t *testing.T) { if event.EventHash == "after-high-water" { t.Fatalf("snapshot included concurrent event: %#v", event) } + if event.RecordDigest == "" { + t.Fatalf("event %d is missing record digest: %#v", index, event) + } } if events[0].TimestampMS != events[1].TimestampMS || events[0].Model != events[1].Model || events[0].Endpoint != events[1].Endpoint || events[0].EventID == events[1].EventID { @@ -125,7 +128,7 @@ func TestUsageSnapshotPagesAllRowsBeyondLegacy50000Limit(t *testing.T) { t.Fatalf("snapshot = %#v", snapshot) } - hash := sha256.New() + digest := initialSnapshotDigest() var delivered int64 var afterID int64 for { @@ -134,7 +137,7 @@ func TestUsageSnapshotPagesAllRowsBeyondLegacy50000Limit(t *testing.T) { t.Fatalf("read page after %d: %v", afterID, err) } for _, event := range page.Events { - _, _ = fmt.Fprintf(hash, "%d\x00%s\n", event.EventID, event.EventHash) + digest = extendSnapshotDigest(digest, event) afterID = event.EventID delivered++ } @@ -145,7 +148,7 @@ func TestUsageSnapshotPagesAllRowsBeyondLegacy50000Limit(t *testing.T) { if delivered != rowCount || afterID != rowCount { t.Fatalf("delivered = %d, last id = %d", delivered, afterID) } - if got := "sha256:" + hex.EncodeToString(hash.Sum(nil)); got != snapshot.Digest { + if got := "sha256:" + hex.EncodeToString(digest); got != snapshot.Digest { t.Fatalf("digest = %q, want %q", got, snapshot.Digest) } } @@ -168,9 +171,21 @@ func snapshotTestEvent(hash string, timestampMS int64) usage.Event { } func snapshotDigest(events []SnapshotEvent) string { - hash := sha256.New() + digest := initialSnapshotDigest() for _, event := range events { - _, _ = fmt.Fprintf(hash, "%d\x00%s\n", event.EventID, event.EventHash) + digest = extendSnapshotDigest(digest, event) } - return "sha256:" + hex.EncodeToString(hash.Sum(nil)) + return "sha256:" + hex.EncodeToString(digest) +} + +func initialSnapshotDigest() []byte { + seed := sha256.Sum256([]byte("cpa-manager-plus:usage-snapshot:v1")) + return append([]byte(nil), seed[:]...) +} + +func extendSnapshotDigest(previous []byte, event SnapshotEvent) []byte { + hash := sha256.New() + _, _ = hash.Write(previous) + _, _ = fmt.Fprintf(hash, "%d\x00%s\n", event.EventID, event.RecordDigest) + return hash.Sum(nil) } diff --git a/apps/manager-server/internal/service/usage/snapshot_test.go b/apps/manager-server/internal/service/usage/snapshot_test.go index 73530f1e5..703601a6c 100644 --- a/apps/manager-server/internal/service/usage/snapshot_test.go +++ b/apps/manager-server/internal/service/usage/snapshot_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + sqliterepo "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/sqlite" "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/store" usageparser "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/usage" ) @@ -87,6 +88,51 @@ func TestSnapshotProtocolReplaysAndRejectsTamperedCrossSnapshotCursor(t *testing requireSnapshotErrorCode(t, err, SnapshotErrorInvalidCursor) } +func TestSnapshotProtocolFailsClosedWhenFrozenRowsChange(t *testing.T) { + now := time.Unix(1_800_000_000, 0) + path := filepath.Join(t.TempDir(), "usage.sqlite") + db, err := store.Open(path) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if _, err := db.InsertEvents(context.Background(), []usageparser.Event{ + snapshotServiceEvent("event-a", 1), + snapshotServiceEvent("event-b", 2), + }); err != nil { + t.Fatalf("insert events: %v", err) + } + + service := New(db, WithSnapshotProtocol(SnapshotProtocolConfig{ + SigningKey: []byte("0123456789abcdef0123456789abcdef"), + TTL: time.Hour, + Now: func() time.Time { return now }, + })) + first, err := service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{Limit: 1}) + if err != nil { + t.Fatalf("read first page: %v", err) + } + + raw, err := sqliterepo.Open(path) + if err != nil { + t.Fatalf("open raw database: %v", err) + } + if _, err := raw.Exec(`update usage_events set model = 'mutated' where id = 2`); err != nil { + _ = raw.Close() + t.Fatalf("mutate frozen row: %v", err) + } + if err := raw.Close(); err != nil { + t.Fatalf("close raw database: %v", err) + } + + _, err = service.ReadSnapshotPage(context.Background(), SnapshotPageRequest{ + SnapshotID: first.SnapshotID, + Cursor: first.NextCursor, + Limit: 1, + }) + requireSnapshotErrorCode(t, err, SnapshotErrorChanged) +} + func TestSnapshotProtocolExpiresSnapshotAndCursor(t *testing.T) { now := time.Unix(1_800_000_000, 0) db, err := store.Open(filepath.Join(t.TempDir(), "usage.sqlite"))