| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "testing" |
| 10 | |
| 11 | "reasonix/internal/provider" |
| 12 | ) |
| 13 | |
| 14 | func TestMessageRetractStrictSchema(t *testing.T) { |
| 15 | for _, payload := range []string{`{}`, `{"messageIds":null}`, `{"messageIds":[]}`, `{"messageIds":[""]}`, `{"messageIds":[" "]}`, `{"messageIds":["a","a"]}`, `{"messageIds":["a"],"typo":true}`, `{"messageIds":"a"}`} { |
| 16 | t.Run(payload, func(t *testing.T) { |
| 17 | _, err := Project([]Commit{{Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(payload)}}}}) |
| 18 | if !errors.Is(err, ErrDamagedStore) { |
| 19 | t.Fatalf("invalid retract payload accepted: %s, err=%v", payload, err) |
| 20 | } |
| 21 | }) |
| 22 | } |
| 23 | } |
| 24 | |
| 25 | func TestMessageRetractProjectionAndRestore(t *testing.T) { |
| 26 | messageEvent := func(kind, id string) Event { |
| 27 | payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: id, Role: provider.RoleUser, Content: id}}) |
| 28 | if err != nil { |
| 29 | t.Fatal(err) |
| 30 | } |
| 31 | return Event{Kind: kind, Payload: payload} |
| 32 | } |
| 33 | commits := []Commit{{Events: []Event{messageEvent("message/complete", "retained"), messageEvent("message/complete", "removed")}}, {Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["removed"],"reason":"synthetic-turn-interrupted"}`)}}}} |
| 34 | check := func(want []string) { |
| 35 | t.Helper() |
| 36 | p, err := Project(commits) |
| 37 | if err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | for name, messages := range map[string][]provider.Message{"history": p.Messages, "model": p.ModelMessages} { |
| 41 | ids := make([]string, len(messages)) |
| 42 | for i := range messages { |
| 43 | ids[i] = messages[i].ID |
| 44 | } |
| 45 | if !idsEqual(ids, want) { |
| 46 | t.Fatalf("%s ids=%v, want=%v", name, ids, want) |
| 47 | } |
| 48 | } |
| 49 | } |
| 50 | check([]string{"retained"}) |
| 51 | commits = append(commits, commits[1]) |
| 52 | check([]string{"retained"}) |
| 53 | commits = append(commits, Commit{Events: []Event{messageEvent("message/upsert", "removed")}}) |
| 54 | check([]string{"retained", "removed"}) |
| 55 | } |
| 56 | |
| 57 | func TestMessageRetractRestoresOriginalTurnAfterReopen(t *testing.T) { |
| 58 | t.Run("checkpoint", func(t *testing.T) { testRetractedTurnRestore(t, false) }) |
| 59 | t.Run("log-replay", func(t *testing.T) { testRetractedTurnRestore(t, true) }) |
| 60 | } |
| 61 | |
| 62 | func testRetractedTurnRestore(t *testing.T, removeCache bool) { |
| 63 | root := filepath.Join(t.TempDir(), "sessions") |
| 64 | service, err := NewService("local", NewFilesystemPersistence(root)) |
| 65 | if err != nil { |
| 66 | t.Fatal(err) |
| 67 | } |
| 68 | t.Cleanup(func() { _ = service.CloseAll(context.Background()) }) |
| 69 | runtime, err := service.Create(t.Context(), CreateOptions{}) |
| 70 | if err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | user := json.RawMessage(`{"message":{"id":"input","role":"user","content":"restored input"}}`) |
| 74 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "original", TurnID: "original-turn", Events: []Event{ |
| 75 | {Kind: "turn/start", Payload: json.RawMessage(`{}`)}, |
| 76 | {Kind: "message/complete", Payload: user}, |
| 77 | {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 78 | }}); err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "retract", Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["input"]}`)}}}); err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | if _, err := runtime.Session().Flush(t.Context()); err != nil { |
| 85 | t.Fatal(err) |
| 86 | } |
| 87 | ref := runtime.Ref() |
| 88 | if err := service.Close(t.Context(), ref); err != nil { |
| 89 | t.Fatal(err) |
| 90 | } |
| 91 | if removeCache { |
| 92 | if err := os.RemoveAll(recoveryCacheDir(filepath.Join(root, ref.SessionID))); err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | } |
| 96 | binding, err := service.EnsureExecution(t.Context(), ref) |
| 97 | if err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | defer binding.Release(t.Context()) |
| 101 | runtime, _ = service.Runtime(ref) |
| 102 | // Repair batches intentionally have no live turn identity. |
| 103 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "restore", Events: []Event{{Kind: "message/upsert", Payload: user}}}); err != nil { |
| 104 | t.Fatal(err) |
| 105 | } |
| 106 | if _, err := runtime.Session().Flush(t.Context()); err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | p := runtime.Session().Snapshot().Projection |
| 110 | if p.HiddenTurns["original-turn"] || len(p.TranscriptInputs) != 1 || p.TranscriptInputs[0].TurnID != "original-turn" { |
| 111 | t.Fatalf("restored input lost its original turn: hidden=%v inputs=%+v", p.HiddenTurns, p.TranscriptInputs) |
| 112 | } |
| 113 | info, err := service.Query().Stat(t.Context(), ref) |
| 114 | if err != nil { |
| 115 | t.Fatal(err) |
| 116 | } |
| 117 | if info.Turns != 1 || info.Preview != "restored input" || runtime.Session().RecentSnapshot().TotalTurns != 1 { |
| 118 | t.Fatalf("restored metadata disagrees: info=%+v recent=%+v", info, runtime.Session().RecentSnapshot()) |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | func TestRecoveredLocalOnlyMessageClearsOnlyItsClosedReplyAnchor(t *testing.T) { |
| 123 | _, _, runtime := newSourceService(t, "recovery-anchor") |
| 124 | appendCompletedTurn(t, runtime, "first", "first-reply") |
| 125 | appendCompletedTurn(t, runtime, "second", "second-reply") |
| 126 | payload, _ := json.Marshal(map[string]any{"message": provider.Message{ID: "second-reply", Role: provider.RoleTool, LocalOnly: true}}) |
| 127 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "repair", Events: []Event{{Kind: "message/upsert", Payload: payload}}}); err != nil { |
| 128 | t.Fatal(err) |
| 129 | } |
| 130 | p := runtime.Session().Snapshot().Projection |
| 131 | if p.Turns[0].MessageID != "first-reply" || p.Turns[1].MessageID != "" { |
| 132 | t.Fatalf("invalid recovery anchors: %+v", p.Turns) |
| 133 | } |
| 134 | } |
| 135 | |
| 136 | func TestMessageRetractHistorySearchAndRestart(t *testing.T) { |
| 137 | for _, incremental := range []bool{false, true} { |
| 138 | name := "rebuild" |
| 139 | if incremental { |
| 140 | name = "incremental" |
| 141 | } |
| 142 | t.Run(name, func(t *testing.T) { |
| 143 | root := filepath.Join(t.TempDir(), "sessions") |
| 144 | service, err := NewService("local", NewFilesystemPersistence(root)) |
| 145 | if err != nil { |
| 146 | t.Fatal(err) |
| 147 | } |
| 148 | t.Cleanup(func() { _ = service.CloseAll(context.Background()) }) |
| 149 | runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "retracted"}) |
| 150 | if err != nil { |
| 151 | t.Fatal(err) |
| 152 | } |
| 153 | appendWindowMessages(t, runtime, "retained", "removed", "tail") |
| 154 | query, ref := service.Query(), runtime.Ref() |
| 155 | read := func(req HistoryWindowRequest) HistoryWindowPage { |
| 156 | t.Helper() |
| 157 | if _, _, err := query.prepareHistoryIndex(t.Context(), ref); err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | page, err := query.ReadHistoryWindow(t.Context(), ref, req) |
| 161 | if err != nil || page.Status != "ready" { |
| 162 | t.Fatalf("history: %+v, %v", page, err) |
| 163 | } |
| 164 | return page |
| 165 | } |
| 166 | var oldCursor string |
| 167 | if incremental { |
| 168 | oldCursor = read(HistoryWindowRequest{Anchor: "newest", Limit: 1}).OlderCursor |
| 169 | if page := searchHistoryReady(t, query, ref, "body", "", 10); len(page.Hits) != 3 { |
| 170 | t.Fatalf("initial search: %+v", page) |
| 171 | } |
| 172 | } |
| 173 | appendEvent := func(op, kind, payload string) { |
| 174 | t.Helper() |
| 175 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: op, Events: []Event{{Kind: kind, Payload: json.RawMessage(payload)}}}); err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | if _, err := runtime.Session().Flush(t.Context()); err != nil { |
| 179 | t.Fatal(err) |
| 180 | } |
| 181 | } |
| 182 | appendEvent("retract", "message/retract", `{"messageIds":["removed"],"reason":"synthetic-turn-interrupted"}`) |
| 183 | page := read(HistoryWindowRequest{Anchor: "newest"}) |
| 184 | if !idsEqual(windowIDs(t, page), []string{"retained", "tail"}) { |
| 185 | t.Fatalf("retracted ids=%v", windowIDs(t, page)) |
| 186 | } |
| 187 | if found := searchHistoryReady(t, query, ref, "removed", "", 10); len(found.Hits) != 0 { |
| 188 | t.Fatalf("retracted search: %+v", found) |
| 189 | } |
| 190 | if incremental { |
| 191 | old := read(HistoryWindowRequest{Anchor: "cursor", Cursor: oldCursor, Limit: 10}) |
| 192 | if !idsEqual(windowIDs(t, old), []string{"retained", "removed"}) { |
| 193 | t.Fatalf("fixed snapshot changed: %v", windowIDs(t, old)) |
| 194 | } |
| 195 | } |
| 196 | appendEvent("retract-again", "message/retract", `{"messageIds":["removed"]}`) |
| 197 | read(HistoryWindowRequest{Anchor: "newest"}) |
| 198 | appendEvent("restore", "message/upsert", `{"message":{"id":"removed","role":"user","content":"restored body-removed"}}`) |
| 199 | checkRestored := func() { |
| 200 | t.Helper() |
| 201 | page := read(HistoryWindowRequest{Anchor: "newest"}) |
| 202 | if !idsEqual(windowIDs(t, page), []string{"retained", "tail", "removed"}) { |
| 203 | t.Fatalf("restored ids=%v", windowIDs(t, page)) |
| 204 | } |
| 205 | if page.Messages[2].Version != 2 { |
| 206 | t.Fatalf("restored version=%d, want=2", page.Messages[2].Version) |
| 207 | } |
| 208 | if found := searchHistoryReady(t, query, ref, "restored", "", 10); len(found.Hits) != 1 { |
| 209 | t.Fatalf("restored search: %+v", found) |
| 210 | } |
| 211 | } |
| 212 | checkRestored() |
| 213 | if err := service.CloseAll(t.Context()); err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | for _, removeCache := range []bool{false, true} { |
| 217 | if removeCache { |
| 218 | if err := os.RemoveAll(filepath.Join(root, ".query-cache")); err != nil { |
| 219 | t.Fatal(err) |
| 220 | } |
| 221 | } |
| 222 | reopened, err := NewService("local", NewFilesystemPersistence(root)) |
| 223 | if err != nil { |
| 224 | t.Fatal(err) |
| 225 | } |
| 226 | query = reopened.Query() |
| 227 | checkRestored() |
| 228 | if err := reopened.CloseAll(t.Context()); err != nil { |
| 229 | t.Fatal(err) |
| 230 | } |
| 231 | } |
| 232 | }) |
| 233 | } |
| 234 | } |
| 235 | |
| 236 | func TestMessageRetractRenumbersTurnsWithoutChangingOldSnapshot(t *testing.T) { |
| 237 | for _, removeID := range []string{"u1", "u2"} { |
| 238 | for _, withAnswer := range []bool{false, true} { |
| 239 | name := removeID |
| 240 | if withAnswer { |
| 241 | name += "-with-answer" |
| 242 | } |
| 243 | t.Run(name, func(t *testing.T) { |
| 244 | service, err := NewService("local", NewFilesystemPersistence(filepath.Join(t.TempDir(), "sessions"))) |
| 245 | if err != nil { |
| 246 | t.Fatal(err) |
| 247 | } |
| 248 | t.Cleanup(func() { _ = service.CloseAll(context.Background()) }) |
| 249 | runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "turn-renumber"}) |
| 250 | if err != nil { |
| 251 | t.Fatal(err) |
| 252 | } |
| 253 | appendMessage := func(op, kind, id string, role provider.Role) { |
| 254 | t.Helper() |
| 255 | payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: id, Role: role, Content: "body-" + id}}) |
| 256 | if err != nil { |
| 257 | t.Fatal(err) |
| 258 | } |
| 259 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: op, Events: []Event{{Kind: kind, Payload: payload}}}); err != nil { |
| 260 | t.Fatal(err) |
| 261 | } |
| 262 | } |
| 263 | for _, id := range []string{"u1", "a1", "u2", "a2", "u3", "a3"} { |
| 264 | role := provider.RoleUser |
| 265 | if id[0] == 'a' { |
| 266 | role = provider.RoleAssistant |
| 267 | } |
| 268 | appendMessage(id, "message/complete", id, role) |
| 269 | } |
| 270 | read := func(req HistoryWindowRequest) HistoryWindowPage { |
| 271 | t.Helper() |
| 272 | if _, err := runtime.Session().Flush(t.Context()); err != nil { |
| 273 | t.Fatal(err) |
| 274 | } |
| 275 | if _, _, err := service.Query().prepareHistoryIndex(t.Context(), runtime.Ref()); err != nil { |
| 276 | t.Fatal(err) |
| 277 | } |
| 278 | page, err := service.Query().ReadHistoryWindow(t.Context(), runtime.Ref(), req) |
| 279 | if err != nil || page.Status != "ready" { |
| 280 | t.Fatalf("page=%+v err=%v", page, err) |
| 281 | } |
| 282 | return page |
| 283 | } |
| 284 | old := read(HistoryWindowRequest{Anchor: "newest", Limit: 1}) |
| 285 | if old.TotalTurns != 3 || old.Messages[0].VisibleTurn != 3 { |
| 286 | t.Fatalf("initial turns=%+v", old) |
| 287 | } |
| 288 | ids := []string{removeID} |
| 289 | if withAnswer { |
| 290 | ids = append(ids, "a"+removeID[1:]) |
| 291 | } |
| 292 | payload, _ := json.Marshal(map[string]any{"messageIds": ids}) |
| 293 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "retract-user", Events: []Event{{Kind: "message/retract", Payload: payload}}}); err != nil { |
| 294 | t.Fatal(err) |
| 295 | } |
| 296 | page := read(HistoryWindowRequest{Anchor: "newest"}) |
| 297 | if page.TotalTurns != 2 { |
| 298 | t.Fatalf("current total turns=%d want=2", page.TotalTurns) |
| 299 | } |
| 300 | turn := 0 |
| 301 | for _, m := range page.Messages { |
| 302 | if m.Role == string(provider.RoleUser) { |
| 303 | turn++ |
| 304 | } |
| 305 | if m.VisibleTurn != turn { |
| 306 | t.Fatalf("message %s turn=%d want=%d", m.MessageID, m.VisibleTurn, turn) |
| 307 | } |
| 308 | } |
| 309 | past := read(HistoryWindowRequest{Anchor: "cursor", Cursor: old.OlderCursor, Limit: 10}) |
| 310 | if past.TotalTurns != 3 || !idsEqual(windowIDs(t, past), []string{"u1", "a1", "u2", "a2", "u3"}) { |
| 311 | t.Fatalf("old snapshot changed: %+v", past) |
| 312 | } |
| 313 | for i, m := range past.Messages { |
| 314 | if m.VisibleTurn != i/2+1 { |
| 315 | t.Fatalf("old snapshot %s turn=%d", m.MessageID, m.VisibleTurn) |
| 316 | } |
| 317 | } |
| 318 | appendMessage("restore-user", "message/upsert", removeID, provider.RoleUser) |
| 319 | restored := read(HistoryWindowRequest{Anchor: "newest"}) |
| 320 | last := restored.Messages[len(restored.Messages)-1] |
| 321 | if restored.TotalTurns != 3 || last.MessageID != removeID || last.VisibleTurn != 3 || last.Version != 2 { |
| 322 | t.Fatalf("restored user=%+v total=%d", last, restored.TotalTurns) |
| 323 | } |
| 324 | }) |
| 325 | } |
| 326 | } |
| 327 | } |
| 328 | |
| 329 | func TestMessageRetractCatalogAndRecentFollowVisibleHistory(t *testing.T) { |
| 330 | root := filepath.Join(t.TempDir(), "sessions") |
| 331 | service, err := NewService("local", NewFilesystemPersistence(root)) |
| 332 | if err != nil { |
| 333 | t.Fatal(err) |
| 334 | } |
| 335 | t.Cleanup(func() { _ = service.CloseAll(context.Background()) }) |
| 336 | runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "recent-retraction"}) |
| 337 | if err != nil { |
| 338 | t.Fatal(err) |
| 339 | } |
| 340 | for _, id := range []string{"first", "second"} { |
| 341 | user, _ := json.Marshal(map[string]any{"message": provider.Message{ID: id, Role: provider.RoleUser, Content: id + " question"}}) |
| 342 | answer, _ := json.Marshal(map[string]any{"message": provider.Message{ID: id + "-answer", Role: provider.RoleAssistant, Content: id + " answer"}}) |
| 343 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: id, TurnID: id + "-turn", Events: []Event{{Kind: "turn/start", Payload: json.RawMessage(`{}`)}, {Kind: "message/complete", Payload: user}, {Kind: "message/complete", Payload: answer}, {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}}}); err != nil { |
| 344 | t.Fatal(err) |
| 345 | } |
| 346 | } |
| 347 | if _, err := runtime.Session().Flush(t.Context()); err != nil { |
| 348 | t.Fatal(err) |
| 349 | } |
| 350 | if got := runtime.Session().RecentSnapshot(); got.TotalTurns != 2 { |
| 351 | t.Fatalf("initial recent turns=%d", got.TotalTurns) |
| 352 | } |
| 353 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "remove-first", Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["first","first-answer"]}`)}}}); err != nil { |
| 354 | t.Fatal(err) |
| 355 | } |
| 356 | if _, err := runtime.Session().Flush(t.Context()); err != nil { |
| 357 | t.Fatal(err) |
| 358 | } |
| 359 | check := func(label string, recent RecentSnapshot, info SessionInfo) { |
| 360 | t.Helper() |
| 361 | if recent.TotalTurns != 1 { |
| 362 | t.Errorf("%s recent total=%d want=1", label, recent.TotalTurns) |
| 363 | } |
| 364 | if info.Turns != 1 || info.Preview != "second question" { |
| 365 | t.Errorf("%s catalog turns=%d preview=%q", label, info.Turns, info.Preview) |
| 366 | } |
| 367 | ids := make([]string, 0, len(recent.Entries)) |
| 368 | for _, m := range recent.Entries { |
| 369 | ids = append(ids, m.MessageID) |
| 370 | if m.VisibleTurn != 1 { |
| 371 | t.Errorf("%s recent %s visible turn=%d", label, m.MessageID, m.VisibleTurn) |
| 372 | } |
| 373 | } |
| 374 | if !idsEqual(ids, []string{"second", "second-answer"}) { |
| 375 | t.Errorf("%s recent ids=%v", label, ids) |
| 376 | } |
| 377 | } |
| 378 | info, err := service.Query().Stat(t.Context(), runtime.Ref()) |
| 379 | if err != nil { |
| 380 | t.Fatal(err) |
| 381 | } |
| 382 | check("live", runtime.Session().RecentSnapshot(), info) |
| 383 | targets := ForkTargets(runtime.Session().Snapshot().Projection) |
| 384 | if len(targets.Targets) != 1 || targets.Targets[0].TurnID != "second-turn" || targets.Targets[0].TurnNumber != 1 || targets.Targets[0].MessageID != "second-answer" { |
| 385 | t.Fatalf("withdrawn turn remained forkable or numbering changed: %+v", targets) |
| 386 | } |
| 387 | if seq, reason, err := ForkSequence(runtime.Session().Snapshot().Projection, "first-turn"); err != nil || seq != 0 || reason != ForkHistoryUnverifiable { |
| 388 | t.Fatalf("withdrawn turn fork: seq=%d reason=%s err=%v", seq, reason, err) |
| 389 | } |
| 390 | if seq, reason, err := ForkSequence(runtime.Session().Snapshot().Projection, "second-turn"); err != nil || seq == 0 || reason != ForkAvailable { |
| 391 | t.Fatalf("remaining turn fork: seq=%d reason=%s err=%v", seq, reason, err) |
| 392 | } |
| 393 | ref := runtime.Ref() |
| 394 | if err := service.Close(t.Context(), ref); err != nil { |
| 395 | t.Fatal(err) |
| 396 | } |
| 397 | view, err := service.OpenSession(t.Context(), ref) |
| 398 | if err != nil { |
| 399 | t.Fatal(err) |
| 400 | } |
| 401 | info, err = service.Query().Stat(t.Context(), ref) |
| 402 | if err != nil { |
| 403 | t.Fatal(err) |
| 404 | } |
| 405 | check("cold", view.Recent, info) |
| 406 | // Reopen a writer and remove every remaining input. Neither a cached |
| 407 | // first-preview fallback nor old turn boundaries may resurrect the list |
| 408 | // preview or leave a phantom visible turn. |
| 409 | binding, err := service.EnsureExecution(t.Context(), ref) |
| 410 | if err != nil { |
| 411 | t.Fatal(err) |
| 412 | } |
| 413 | live, ok := service.Runtime(ref) |
| 414 | if !ok { |
| 415 | t.Fatal("reopened runtime missing") |
| 416 | } |
| 417 | if _, err := live.Session().Append(t.Context(), Batch{OperationID: "remove-all-inputs", Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["second","second-answer"]}`)}}}); err != nil { |
| 418 | t.Fatal(err) |
| 419 | } |
| 420 | if _, err := live.Session().Flush(t.Context()); err != nil { |
| 421 | t.Fatal(err) |
| 422 | } |
| 423 | if err := binding.Release(t.Context()); err != nil { |
| 424 | t.Fatal(err) |
| 425 | } |
| 426 | checkEmpty := func(label string, recent RecentSnapshot, info SessionInfo) { |
| 427 | t.Helper() |
| 428 | if recent.TotalTurns != 0 || len(recent.Entries) != 0 { |
| 429 | t.Errorf("%s recent resurrected empty history: %+v", label, recent) |
| 430 | } |
| 431 | if info.Turns != 0 || info.Preview != "" { |
| 432 | t.Errorf("%s catalog resurrected empty history: turns=%d preview=%q", label, info.Turns, info.Preview) |
| 433 | } |
| 434 | } |
| 435 | info, err = service.Query().Stat(t.Context(), ref) |
| 436 | if err != nil { |
| 437 | t.Fatal(err) |
| 438 | } |
| 439 | checkEmpty("empty-live", live.Session().RecentSnapshot(), info) |
| 440 | checkNoForks := func(projection Projection) { |
| 441 | t.Helper() |
| 442 | if targets := ForkTargets(projection); len(targets.Targets) != 0 { |
| 443 | t.Fatalf("empty history exposes fork targets: %+v", targets) |
| 444 | } |
| 445 | for _, id := range []string{"first-turn", "second-turn"} { |
| 446 | if seq, reason, err := ForkSequence(projection, id); err != nil || seq != 0 || reason != ForkHistoryUnverifiable { |
| 447 | t.Fatalf("empty history fork %s: seq=%d reason=%s err=%v", id, seq, reason, err) |
| 448 | } |
| 449 | } |
| 450 | } |
| 451 | checkNoForks(live.Session().Snapshot().Projection) |
| 452 | if err := service.Close(t.Context(), ref); err != nil { |
| 453 | t.Fatal(err) |
| 454 | } |
| 455 | for _, removeCache := range []bool{false, true} { |
| 456 | label := "empty-cached" |
| 457 | if removeCache { |
| 458 | label = "empty-rebuilt" |
| 459 | for _, dir := range []string{filepath.Join(root, ".query-cache", ref.SessionID), recoveryCacheDir(filepath.Join(root, ref.SessionID))} { |
| 460 | if err := os.RemoveAll(dir); err != nil { |
| 461 | t.Fatal(err) |
| 462 | } |
| 463 | } |
| 464 | } |
| 465 | // EnsureExecution deterministically completes checkpoint rebuild rather |
| 466 | // than racing a cold reader's background preparation notification. |
| 467 | binding, err := service.EnsureExecution(t.Context(), ref) |
| 468 | if err != nil { |
| 469 | t.Fatal(err) |
| 470 | } |
| 471 | reopened, ok := service.Runtime(ref) |
| 472 | if !ok { |
| 473 | t.Fatal("reopened runtime missing") |
| 474 | } |
| 475 | info, err := service.Query().Stat(t.Context(), ref) |
| 476 | if err != nil { |
| 477 | t.Fatal(err) |
| 478 | } |
| 479 | checkEmpty(label, reopened.Session().RecentSnapshot(), info) |
| 480 | checkNoForks(reopened.Session().Snapshot().Projection) |
| 481 | if err := binding.Release(t.Context()); err != nil { |
| 482 | t.Fatal(err) |
| 483 | } |
| 484 | if err := service.Close(t.Context(), ref); err != nil { |
| 485 | t.Fatal(err) |
| 486 | } |
| 487 | cold, err := service.OpenSession(t.Context(), ref) |
| 488 | if err != nil { |
| 489 | t.Fatal(err) |
| 490 | } |
| 491 | info, err = service.Query().Stat(t.Context(), ref) |
| 492 | if err != nil { |
| 493 | t.Fatal(err) |
| 494 | } |
| 495 | checkEmpty(label+"-cold", cold.Recent, info) |
| 496 | } |
| 497 | } |
| 498 |