| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "sync/atomic" |
| 10 | "testing" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | type catalogOnlyPersistence struct{ opens atomic.Int32 } |
| 15 | |
| 16 | func (*catalogOnlyPersistence) Create(CreateOptions) (*Session, error) { |
| 17 | return nil, errors.New("unexpected create") |
| 18 | } |
| 19 | func (p *catalogOnlyPersistence) Open(string, AccessMode) (*Session, error) { |
| 20 | p.opens.Add(1) |
| 21 | return nil, errors.New("unexpected history open") |
| 22 | } |
| 23 | func (*catalogOnlyPersistence) Stat(context.Context, string) (SessionInfo, error) { |
| 24 | return SessionInfo{}, errors.New("unexpected stat") |
| 25 | } |
| 26 | func (*catalogOnlyPersistence) List(context.Context, string, int) (SessionPage, error) { |
| 27 | return SessionPage{Sessions: []SessionInfo{{ |
| 28 | SessionID: "listed", Codec: Codec, Title: "cached", Turns: 3, |
| 29 | MetadataStatus: MetadataReady, |
| 30 | }}}, nil |
| 31 | } |
| 32 | |
| 33 | func TestListDoesNotOpenSessionHistoryForReadyMetadata(t *testing.T) { |
| 34 | persistence := &catalogOnlyPersistence{} |
| 35 | query := newQuery("local", persistence, nil) |
| 36 | page, err := query.List(t.Context(), "", 50) |
| 37 | if err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | if len(page.Sessions) != 1 || page.Sessions[0].Title != "cached" || page.Sessions[0].Turns != 3 { |
| 41 | t.Fatalf("page = %+v", page) |
| 42 | } |
| 43 | if got := persistence.opens.Load(); got != 0 { |
| 44 | t.Fatalf("catalog listing opened %d full histories", got) |
| 45 | } |
| 46 | } |
| 47 | |
| 48 | func TestResolveSessionIDReturnsCatalogOwnedIdentity(t *testing.T) { |
| 49 | persistence := &catalogOnlyPersistence{} |
| 50 | query := newQuery("local", persistence, nil) |
| 51 | ref, err := query.ResolveSessionID(t.Context(), "listed") |
| 52 | if err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | if ref != (SessionRef{HostID: "local", SessionID: "listed"}) { |
| 56 | t.Fatalf("resolved ref = %+v", ref) |
| 57 | } |
| 58 | if _, err = query.ResolveSessionID(t.Context(), "../listed"); err == nil { |
| 59 | t.Fatal("path-like session identity was accepted") |
| 60 | } |
| 61 | if _, err = query.ResolveSessionID(t.Context(), "missing"); !errors.Is(err, ErrSessionNotFound) { |
| 62 | t.Fatalf("missing identity error = %v", err) |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | func TestCatalogMetadataIsBoundToSessionIncarnation(t *testing.T) { |
| 67 | cacheDir := t.TempDir() |
| 68 | first := Manifest{SessionID: "same-id", CreatedAt: time.Unix(1, 0).UTC()} |
| 69 | if err := writeCatalogMetadata(cacheDir, metadataFromProjection(first, 0, Projection{Title: "stale"})); err != nil { |
| 70 | t.Fatal(err) |
| 71 | } |
| 72 | second := Manifest{SessionID: "same-id", CreatedAt: time.Unix(2, 0).UTC()} |
| 73 | if _, err := readCatalogMetadata(cacheDir, second, logRevision{}); err == nil { |
| 74 | t.Fatal("metadata from a deleted session was reused by a new incarnation") |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | func TestCatalogMetadataIsBoundToLogRevision(t *testing.T) { |
| 79 | cacheDir := t.TempDir() |
| 80 | manifest := Manifest{SessionID: "rev", CreatedAt: time.Unix(1, 0).UTC()} |
| 81 | written := metadataFromProjection(manifest, 4, Projection{Title: "cached"}) |
| 82 | written.LogSize, written.LogModTimeNS, written.LogIdentity = 128, 7, "digest" |
| 83 | if err := writeCatalogMetadata(cacheDir, written); err != nil { |
| 84 | t.Fatal(err) |
| 85 | } |
| 86 | metadata, err := readCatalogMetadata(cacheDir, manifest, logRevision{Size: 128, ModTimeNS: 7, Identity: "digest", Exists: true}) |
| 87 | if err != nil || metadata.Title != "cached" || metadata.Sequence != 4 { |
| 88 | t.Fatalf("matching revision rejected: %+v, %v", metadata, err) |
| 89 | } |
| 90 | // A log that grew by even one byte invalidates the projection without any |
| 91 | // need to replay it. |
| 92 | if _, err := readCatalogMetadata(cacheDir, manifest, logRevision{Size: 129, ModTimeNS: 7, Identity: "digest", Exists: true}); err == nil { |
| 93 | t.Fatal("stale log revision was accepted") |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | type pagedCatalogHandle struct{ reads int } |
| 98 | |
| 99 | func (h *pagedCatalogHandle) Read(_ context.Context, offset uint64, _ int) (EventPage, error) { |
| 100 | h.reads++ |
| 101 | if offset == 0 { |
| 102 | return EventPage{Commits: []Commit{{FirstSequence: 1, EventCount: 1, Events: []Event{{Kind: "diagnostic", Optional: true, Sequence: 1}}}}, Next: 1, Truncated: true}, nil |
| 103 | } |
| 104 | return EventPage{Commits: []Commit{{FirstSequence: 2, EventCount: 1, Events: []Event{{Kind: "diagnostic", Optional: true, Sequence: 2}}}}, Next: 2}, nil |
| 105 | } |
| 106 | func (*pagedCatalogHandle) Append(context.Context, Batch) (Commit, error) { |
| 107 | return Commit{}, ErrReadOnly |
| 108 | } |
| 109 | func (*pagedCatalogHandle) Flush(context.Context) (DurableReceipt, error) { |
| 110 | return DurableReceipt{}, nil |
| 111 | } |
| 112 | func (*pagedCatalogHandle) Close(context.Context) error { return nil } |
| 113 | |
| 114 | func TestCatalogMetadataRebuildAdvancesPagedCursor(t *testing.T) { |
| 115 | handle := &pagedCatalogHandle{} |
| 116 | cacheDir := filepath.Join(t.TempDir(), "cache") |
| 117 | manifest := Manifest{SessionID: "paged", CreatedAt: time.Unix(1, 0).UTC()} |
| 118 | if err := rebuildCatalogMetadata(t.Context(), handle, cacheDir, t.TempDir(), manifest); err != nil { |
| 119 | t.Fatal(err) |
| 120 | } |
| 121 | if handle.reads != 2 { |
| 122 | t.Fatalf("reads = %d, want 2", handle.reads) |
| 123 | } |
| 124 | if _, err := os.Stat(catalogMetadataPath(cacheDir)); err != nil { |
| 125 | t.Fatal(err) |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | // TestWarmListDoesNotReplayEventBodies proves catalog listing never falls back |
| 130 | // to a log scan. Building the sparse offset index is the observable witness: it |
| 131 | // only happens when a caller replays events, so its absence after List shows the |
| 132 | // page was served from the manifest head, the log revision, and the metadata |
| 133 | // cache alone. |
| 134 | func TestWarmListDoesNotReplayEventBodies(t *testing.T) { |
| 135 | root := filepath.Join(t.TempDir(), "sessions-v4") |
| 136 | persistence := NewFilesystemPersistence(root) |
| 137 | session, err := persistence.Create(CreateOptions{SessionID: "warm"}) |
| 138 | if err != nil { |
| 139 | t.Fatal(err) |
| 140 | } |
| 141 | payload, _ := json.Marshal(map[string]any{"title": "Warm title"}) |
| 142 | if _, err := session.AppendBatch(t.Context(), "title", []Event{{Kind: "session/title", Payload: payload}}); err != nil { |
| 143 | t.Fatal(err) |
| 144 | } |
| 145 | if _, err := session.Append(t.Context(), Batch{OperationID: "turn", TurnID: "turn-1", Events: []Event{ |
| 146 | {Kind: "turn/start"}, {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 147 | }}); err != nil { |
| 148 | t.Fatal(err) |
| 149 | } |
| 150 | if _, err := session.Flush(t.Context()); err != nil { |
| 151 | t.Fatal(err) |
| 152 | } |
| 153 | if err := session.Close(t.Context()); err != nil { |
| 154 | t.Fatal(err) |
| 155 | } |
| 156 | |
| 157 | cacheDir := filepath.Join(root, ".query-cache", "warm") |
| 158 | indexPath := filepath.Join(cacheDir, "events.offset-index.json") |
| 159 | if err := os.Remove(indexPath); err != nil && !os.IsNotExist(err) { |
| 160 | t.Fatal(err) |
| 161 | } |
| 162 | |
| 163 | service, err := NewService("local", persistence) |
| 164 | if err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | t.Cleanup(func() { _ = service.CloseAll(context.Background()) }) |
| 168 | page, err := service.Query().List(t.Context(), "", 50) |
| 169 | if err != nil { |
| 170 | t.Fatal(err) |
| 171 | } |
| 172 | if len(page.Sessions) != 1 { |
| 173 | t.Fatalf("sessions = %+v", page.Sessions) |
| 174 | } |
| 175 | got := page.Sessions[0] |
| 176 | if got.MetadataStatus != MetadataReady || got.Title != "Warm title" || got.Turns != 1 || got.EventSequence != 3 { |
| 177 | t.Fatalf("warm listing = %+v", got) |
| 178 | } |
| 179 | if _, statErr := os.Stat(indexPath); statErr == nil { |
| 180 | t.Fatal("warm List rebuilt the sparse offset index, so it replayed event bodies") |
| 181 | } else if !os.IsNotExist(statErr) { |
| 182 | t.Fatal(statErr) |
| 183 | } |
| 184 | } |
| 185 |