| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "os" |
| 6 | "path/filepath" |
| 7 | "reflect" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "sync/atomic" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "reasonix/desktop/internal/workspacestate" |
| 15 | "reasonix/internal/config" |
| 16 | "reasonix/internal/identitylock" |
| 17 | "reasonix/internal/provider" |
| 18 | "reasonix/internal/session" |
| 19 | ) |
| 20 | |
| 21 | func newHistoricalLifecycleApp(t *testing.T) *App { |
| 22 | t.Helper() |
| 23 | app := NewApp() |
| 24 | app.ctx = t.Context() |
| 25 | installNoopRuntimeEvents(app) |
| 26 | t.Cleanup(app.closeSessionServices) |
| 27 | t.Cleanup(app.stopHistoricalImports) |
| 28 | return app |
| 29 | } |
| 30 | |
| 31 | func historicalLifecycleID(t *testing.T, app *App, title string) string { |
| 32 | t.Helper() |
| 33 | list, err := app.ListHistoricalSessions() |
| 34 | if err != nil { |
| 35 | t.Fatal(err) |
| 36 | } |
| 37 | for _, view := range list.Items { |
| 38 | if view.Title == title { |
| 39 | return view.ID |
| 40 | } |
| 41 | } |
| 42 | t.Fatalf("historical source %q missing: %+v", title, list) |
| 43 | return "" |
| 44 | } |
| 45 | |
| 46 | func awaitHistoricalBatch(t *testing.T, app *App) HistoricalImportStatus { |
| 47 | t.Helper() |
| 48 | // Start/Resume admits its worker before returning. Join that owner instead |
| 49 | // of launching directory discovery to poll it; discovery is intentionally |
| 50 | // rejected after shutdown and is not evidence that a batch has drained. |
| 51 | app.historicalImports.workers.Wait() |
| 52 | status := app.GetHistoricalImportStatus() |
| 53 | if status.Running { |
| 54 | t.Fatalf("historical worker drained while batch remained running: %+v", status) |
| 55 | } |
| 56 | return status |
| 57 | } |
| 58 | |
| 59 | func TestHistoricalBatchContinuesPastBusySource(t *testing.T) { |
| 60 | isolateDesktopUserDirs(t) |
| 61 | root := config.SessionStoreDir() |
| 62 | coldV4MigrationFixture(t, root, "busy") |
| 63 | coldV4MigrationFixture(t, root, "available") |
| 64 | app := newHistoricalLifecycleApp(t) |
| 65 | busy := historicalLifecycleID(t, app, "busy") |
| 66 | available := historicalLifecycleID(t, app, "available") |
| 67 | release, err := identitylock.Acquire(t.Context(), filepath.Join(root, ".busy.ownership.lock")) |
| 68 | if err != nil { |
| 69 | t.Fatal(err) |
| 70 | } |
| 71 | defer release() |
| 72 | if _, err := app.StartHistoricalImport([]string{busy, available}); err != nil { |
| 73 | t.Fatal(err) |
| 74 | } |
| 75 | status := awaitHistoricalBatch(t, app) |
| 76 | states := map[string]string{} |
| 77 | for _, view := range status.Items { |
| 78 | states[view.ID] = view.Status |
| 79 | } |
| 80 | if states[busy] != "blocked" || states[available] != "imported" { |
| 81 | t.Fatalf("one occupied source prevented independent progress: %+v", status) |
| 82 | } |
| 83 | if !app.runtimeRebuildMu.TryLock() { |
| 84 | t.Fatal("batch retained global runtime gate") |
| 85 | } |
| 86 | app.runtimeRebuildMu.Unlock() |
| 87 | } |
| 88 | |
| 89 | func TestHistoricalConcurrentRequestsKeepOneTarget(t *testing.T) { |
| 90 | isolateDesktopUserDirs(t) |
| 91 | coldV4MigrationFixture(t, config.SessionStoreDir(), "same-source") |
| 92 | app := newHistoricalLifecycleApp(t) |
| 93 | id := historicalLifecycleID(t, app, "same-source") |
| 94 | entered, proceed := make(chan struct{}), make(chan struct{}) |
| 95 | var releaseOnce sync.Once |
| 96 | release := func() { releaseOnce.Do(func() { close(proceed) }) } |
| 97 | t.Cleanup(release) |
| 98 | var commits atomic.Int32 |
| 99 | app.desktopSessions.beforeMigrationRegistryCommit = func() error { |
| 100 | if commits.Add(1) == 1 { |
| 101 | close(entered) |
| 102 | <-proceed |
| 103 | } |
| 104 | return nil |
| 105 | } |
| 106 | type outcome struct { |
| 107 | result SessionRestoreResult |
| 108 | err error |
| 109 | } |
| 110 | results := make(chan outcome, 2) |
| 111 | go func() { result, err := app.ImportHistoricalSession(id); results <- outcome{result, err} }() |
| 112 | select { |
| 113 | case <-entered: |
| 114 | case <-time.After(10 * time.Second): |
| 115 | t.Fatal("first import did not reach publication") |
| 116 | } |
| 117 | started := make(chan struct{}) |
| 118 | go func() { |
| 119 | close(started) |
| 120 | result, err := app.ImportHistoricalSession(id) |
| 121 | results <- outcome{result, err} |
| 122 | }() |
| 123 | <-started |
| 124 | release() |
| 125 | first, second := <-results, <-results |
| 126 | if first.err != nil || second.err != nil || first.result.Session != second.result.Session { |
| 127 | t.Fatalf("duplicate import diverged: %+v / %+v", first, second) |
| 128 | } |
| 129 | if commits.Load() != 1 { |
| 130 | t.Fatalf("duplicate request published %d times", commits.Load()) |
| 131 | } |
| 132 | state, err := app.workspaceRegistry().Load(t.Context()) |
| 133 | if err != nil || len(state.SourceMappings) != 1 || len(state.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs) != 1 { |
| 134 | t.Fatalf("duplicate durable identities: %+v %v", state, err) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | func TestPrepareSessionReturnsRevisionedSharedTask(t *testing.T) { |
| 139 | isolateDesktopUserDirs(t) |
| 140 | coldV4MigrationFixture(t, config.SessionStoreDir(), "prepared-navigation") |
| 141 | app := newHistoricalLifecycleApp(t) |
| 142 | list, err := app.ListHistoricalSessions() |
| 143 | if err != nil || len(list.Items) != 1 || list.Items[0].Source == nil { |
| 144 | t.Fatalf("historical listing: %+v %v", list, err) |
| 145 | } |
| 146 | entered, proceed := make(chan struct{}), make(chan struct{}) |
| 147 | app.desktopSessions.beforeMigrationRegistryCommit = func() error { |
| 148 | select { |
| 149 | case <-entered: |
| 150 | default: |
| 151 | close(entered) |
| 152 | } |
| 153 | <-proceed |
| 154 | return nil |
| 155 | } |
| 156 | first, err := app.PrepareSession(SessionSelector{Source: list.Items[0].Source}) |
| 157 | if err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | second, err := app.PrepareSession(SessionSelector{Source: list.Items[0].Source}) |
| 161 | if err != nil || first.OperationID != second.OperationID { |
| 162 | t.Fatalf("duplicate prepare did not join: %+v %+v %v", first, second, err) |
| 163 | } |
| 164 | select { |
| 165 | case <-entered: |
| 166 | case <-time.After(10 * time.Second): |
| 167 | t.Fatal("preparation did not reach publication") |
| 168 | } |
| 169 | close(proceed) |
| 170 | deadline := time.Now().Add(10 * time.Second) |
| 171 | for { |
| 172 | view, getErr := app.GetSessionPreparation(first.OperationID) |
| 173 | if getErr != nil { |
| 174 | t.Fatal(getErr) |
| 175 | } |
| 176 | if view.Status == "ready" { |
| 177 | if view.Target == nil || view.Revision <= first.Revision { |
| 178 | t.Fatalf("invalid terminal preparation: %+v", view) |
| 179 | } |
| 180 | break |
| 181 | } |
| 182 | if time.Now().After(deadline) { |
| 183 | t.Fatalf("preparation did not complete: %+v", view) |
| 184 | } |
| 185 | time.Sleep(time.Millisecond) |
| 186 | } |
| 187 | } |
| 188 | |
| 189 | func TestHistoricalQueueRestartsPaused(t *testing.T) { |
| 190 | isolateDesktopUserDirs(t) |
| 191 | root := config.SessionStoreDir() |
| 192 | coldV4MigrationFixture(t, root, "queue-one") |
| 193 | coldV4MigrationFixture(t, root, "queue-two") |
| 194 | app := newHistoricalLifecycleApp(t) |
| 195 | first := historicalLifecycleID(t, app, "queue-one") |
| 196 | second := historicalLifecycleID(t, app, "queue-two") |
| 197 | c := &app.historicalImports |
| 198 | c.mu.Lock() |
| 199 | c.queue = []string{first, second} |
| 200 | if err := c.saveQueueLocked(); err != nil { |
| 201 | c.mu.Unlock() |
| 202 | t.Fatal(err) |
| 203 | } |
| 204 | c.mu.Unlock() |
| 205 | app.closeSessionServices() |
| 206 | app = newHistoricalLifecycleApp(t) |
| 207 | status, err := app.ListHistoricalSessions() |
| 208 | if err != nil { |
| 209 | t.Fatal(err) |
| 210 | } |
| 211 | if status.Running || !status.Paused || status.Remaining != 2 { |
| 212 | t.Fatalf("restarted queue must wait for manual continue: %+v", status) |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | func TestHistoricalSourceUpdateImportsOneStableBranch(t *testing.T) { |
| 217 | for _, oldIdentity := range []bool{false, true} { |
| 218 | name := "current" |
| 219 | if oldIdentity { |
| 220 | name = "old-identity" |
| 221 | } |
| 222 | t.Run(name, func(t *testing.T) { testHistoricalSourceUpdateImportsOneStableBranch(t, oldIdentity) }) |
| 223 | } |
| 224 | } |
| 225 | |
| 226 | func testHistoricalSourceUpdateImportsOneStableBranch(t *testing.T, oldIdentity bool) { |
| 227 | isolateDesktopUserDirs(t) |
| 228 | old := coldV4MigrationFixture(t, config.SessionStoreDir(), "updated-source") |
| 229 | app := newHistoricalLifecycleApp(t) |
| 230 | list, err := app.ListHistoricalSessions() |
| 231 | if err != nil || len(list.Items) != 1 || list.Items[0].Source == nil { |
| 232 | t.Fatalf("list: %+v %v", list, err) |
| 233 | } |
| 234 | base, err := app.ImportHistoricalSession(list.Items[0].ID) |
| 235 | if err != nil { |
| 236 | t.Fatal(err) |
| 237 | } |
| 238 | if oldIdentity { |
| 239 | rewriteHistoricalSourceIdentityForTest(t, app, list.Items[0].ID) |
| 240 | } |
| 241 | binding, err := old.Open(t.Context(), session.SessionRef{HostID: "migration-source", SessionID: "updated-source"}) |
| 242 | if err != nil { |
| 243 | t.Fatal(err) |
| 244 | } |
| 245 | payload, _ := json.Marshal(map[string]any{"message": provider.Message{ID: "new", Role: provider.RoleAssistant, Content: "new historical content"}}) |
| 246 | if _, err := binding.Runtime().Session().AppendBatch(t.Context(), "new-content", []session.Event{{Kind: "message/complete", Payload: payload}}); err != nil { |
| 247 | t.Fatal(err) |
| 248 | } |
| 249 | oldRef := binding.Runtime().Ref() |
| 250 | if err := binding.Release(t.Context()); err != nil { |
| 251 | t.Fatal(err) |
| 252 | } |
| 253 | if err := old.Close(t.Context(), oldRef); err != nil { |
| 254 | t.Fatal(err) |
| 255 | } |
| 256 | selector := SessionSelector{Ref: &base.Session} |
| 257 | update, err := app.CheckHistoricalSourceUpdate(selector) |
| 258 | if err != nil || update.Status != "checking" { |
| 259 | t.Fatalf("initial check: %+v %v", update, err) |
| 260 | } |
| 261 | deadline := time.Now().Add(10 * time.Second) |
| 262 | for update.Status == "checking" { |
| 263 | if time.Now().After(deadline) { |
| 264 | t.Fatal("source update check did not complete") |
| 265 | } |
| 266 | time.Sleep(time.Millisecond) |
| 267 | update, err = app.CheckHistoricalSourceUpdate(selector) |
| 268 | if err != nil { |
| 269 | t.Fatal(err) |
| 270 | } |
| 271 | } |
| 272 | if update.Status != "available" || update.Version == "" || update.Source == nil { |
| 273 | t.Fatalf("updated source not detected: %+v", update) |
| 274 | } |
| 275 | prepared, err := app.PrepareHistoricalSourceVersion(*update.Source, update.Version) |
| 276 | if err != nil { |
| 277 | t.Fatal(err) |
| 278 | } |
| 279 | for prepared.Status != "ready" { |
| 280 | if time.Now().After(deadline) { |
| 281 | t.Fatal("updated source preparation did not complete") |
| 282 | } |
| 283 | time.Sleep(time.Millisecond) |
| 284 | prepared, err = app.GetSessionPreparation(prepared.OperationID) |
| 285 | if err != nil { |
| 286 | t.Fatal(err) |
| 287 | } |
| 288 | } |
| 289 | if prepared.Target == nil || prepared.Target.SessionID == base.Session.SessionID { |
| 290 | t.Fatalf("source update did not create an independent branch: base=%+v update=%+v", base, prepared) |
| 291 | } |
| 292 | again, err := app.PrepareHistoricalSourceVersion(*list.Items[0].Source, update.Version) |
| 293 | if err != nil || again.OperationID != prepared.OperationID { |
| 294 | t.Fatalf("same version was not deduplicated: %+v %v", again, err) |
| 295 | } |
| 296 | } |
| 297 | |
| 298 | func TestHistoricalCancelCanRestartDurableImport(t *testing.T) { |
| 299 | isolateDesktopUserDirs(t) |
| 300 | coldV4MigrationFixture(t, config.SessionStoreDir(), "cancelled") |
| 301 | app := newHistoricalLifecycleApp(t) |
| 302 | id := historicalLifecycleID(t, app, "cancelled") |
| 303 | entered, proceed := make(chan struct{}), make(chan struct{}) |
| 304 | var releaseOnce sync.Once |
| 305 | release := func() { releaseOnce.Do(func() { close(proceed) }) } |
| 306 | t.Cleanup(release) |
| 307 | var hooks atomic.Int32 |
| 308 | app.desktopSessions.beforeMigrationRegistryCommit = func() error { |
| 309 | if hooks.Add(1) == 1 { |
| 310 | close(entered) |
| 311 | <-proceed |
| 312 | } |
| 313 | return nil |
| 314 | } |
| 315 | if _, err := app.StartHistoricalImport([]string{id}); err != nil { |
| 316 | t.Fatal(err) |
| 317 | } |
| 318 | select { |
| 319 | case <-entered: |
| 320 | case <-time.After(10 * time.Second): |
| 321 | t.Fatal("import did not reach publication") |
| 322 | } |
| 323 | if _, err := app.ControlHistoricalImport("cancel"); err != nil { |
| 324 | t.Fatal(err) |
| 325 | } |
| 326 | release() |
| 327 | awaitHistoricalBatch(t, app) |
| 328 | if _, err := app.StartHistoricalImport([]string{id}); err != nil { |
| 329 | t.Fatalf("cancel permanently disabled explicit import: %v", err) |
| 330 | } |
| 331 | status := awaitHistoricalBatch(t, app) |
| 332 | if len(status.Items) != 1 || status.Items[0].Status != "imported" { |
| 333 | t.Fatalf("cancelled durable import was not resumed: %+v", status) |
| 334 | } |
| 335 | state, err := app.workspaceRegistry().Load(t.Context()) |
| 336 | if err != nil || len(state.SourceMappings) != 1 || len(state.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs) != 1 { |
| 337 | t.Fatalf("restart duplicated target: %+v %v", state, err) |
| 338 | } |
| 339 | } |
| 340 | |
| 341 | func TestHistoricalLateCancelCannotCancelRetry(t *testing.T) { |
| 342 | isolateDesktopUserDirs(t) |
| 343 | root := config.SessionStoreDir() |
| 344 | coldV4MigrationFixture(t, root, "late-cancel") |
| 345 | app := newHistoricalLifecycleApp(t) |
| 346 | id := historicalLifecycleID(t, app, "late-cancel") |
| 347 | releaseSource, err := identitylock.Acquire(t.Context(), filepath.Join(root, ".late-cancel.ownership.lock")) |
| 348 | if err != nil { |
| 349 | t.Fatal(err) |
| 350 | } |
| 351 | first, err := app.prepareHistoricalSession(id, true, false) |
| 352 | if err != nil { |
| 353 | t.Fatal(err) |
| 354 | } |
| 355 | if _, err = waitHistoricalImport(first); err == nil { |
| 356 | t.Fatal("occupied source did not block the first preparation") |
| 357 | } |
| 358 | releaseSource() |
| 359 | |
| 360 | entered, proceed := make(chan struct{}), make(chan struct{}) |
| 361 | var releaseOnce sync.Once |
| 362 | release := func() { releaseOnce.Do(func() { close(proceed) }) } |
| 363 | t.Cleanup(release) |
| 364 | app.desktopSessions.beforeMigrationRegistryCommit = func() error { |
| 365 | close(entered) |
| 366 | <-proceed |
| 367 | return nil |
| 368 | } |
| 369 | second, err := app.prepareHistoricalSession(id, true, false) |
| 370 | if err != nil { |
| 371 | t.Fatal(err) |
| 372 | } |
| 373 | if first.operationID == second.operationID { |
| 374 | t.Fatalf("retry reused operation id %q", second.operationID) |
| 375 | } |
| 376 | select { |
| 377 | case <-entered: |
| 378 | case <-time.After(10 * time.Second): |
| 379 | t.Fatal("retry did not reach publication") |
| 380 | } |
| 381 | if _, err := app.CancelSessionPreparation(first.operationID); err == nil { |
| 382 | t.Fatal("stale operation id remained cancellable after retry") |
| 383 | } |
| 384 | if err := second.ctx.Err(); err != nil { |
| 385 | t.Fatalf("stale cancellation reached retry: %v", err) |
| 386 | } |
| 387 | release() |
| 388 | if result, err := waitHistoricalImport(second); err != nil || result.Session.SessionID == "" { |
| 389 | t.Fatalf("retry did not complete after stale cancellation: %+v %v", result, err) |
| 390 | } |
| 391 | } |
| 392 | |
| 393 | func TestHistoricalCancelledOperationRetryGetsNewIdentity(t *testing.T) { |
| 394 | isolateDesktopUserDirs(t) |
| 395 | coldV4MigrationFixture(t, config.SessionStoreDir(), "cancel-retry") |
| 396 | app := newHistoricalLifecycleApp(t) |
| 397 | id := historicalLifecycleID(t, app, "cancel-retry") |
| 398 | entered, proceed := make(chan struct{}), make(chan struct{}) |
| 399 | var hookCalls atomic.Int32 |
| 400 | app.desktopSessions.beforeMigrationRegistryCommit = func() error { |
| 401 | if hookCalls.Add(1) == 1 { |
| 402 | close(entered) |
| 403 | <-proceed |
| 404 | } |
| 405 | return nil |
| 406 | } |
| 407 | first, err := app.prepareHistoricalSession(id, true, false) |
| 408 | if err != nil { |
| 409 | t.Fatal(err) |
| 410 | } |
| 411 | select { |
| 412 | case <-entered: |
| 413 | case <-time.After(10 * time.Second): |
| 414 | t.Fatal("first preparation did not reach the cancellation point") |
| 415 | } |
| 416 | if _, err = app.CancelSessionPreparation(first.operationID); err != nil { |
| 417 | t.Fatal(err) |
| 418 | } |
| 419 | close(proceed) |
| 420 | if _, err = waitHistoricalImport(first); err == nil { |
| 421 | t.Fatal("cancelled preparation completed successfully") |
| 422 | } |
| 423 | |
| 424 | second, err := app.prepareHistoricalSession(id, true, false) |
| 425 | if err != nil { |
| 426 | t.Fatal(err) |
| 427 | } |
| 428 | if second.operationID == first.operationID { |
| 429 | t.Fatalf("retry reused cancelled operation id %q", second.operationID) |
| 430 | } |
| 431 | result, err := waitHistoricalImport(second) |
| 432 | if err != nil || result.Session.SessionID == "" { |
| 433 | t.Fatalf("retry did not complete: %+v %v", result, err) |
| 434 | } |
| 435 | if _, err = app.GetSessionPreparation(first.operationID); err == nil { |
| 436 | t.Fatal("cancelled operation remained addressable after retry") |
| 437 | } |
| 438 | state, err := app.workspaceRegistry().Load(t.Context()) |
| 439 | if err != nil || len(state.SourceMappings) != 1 || len(state.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs) != 1 { |
| 440 | t.Fatalf("cancel retry duplicated the durable target: %+v %v", state, err) |
| 441 | } |
| 442 | } |
| 443 | |
| 444 | func TestHistoricalInteractiveCancelPreservesBatchDemand(t *testing.T) { |
| 445 | isolateDesktopUserDirs(t) |
| 446 | coldV4MigrationFixture(t, config.SessionStoreDir(), "shared-demand") |
| 447 | app := newHistoricalLifecycleApp(t) |
| 448 | id := historicalLifecycleID(t, app, "shared-demand") |
| 449 | entered, proceed := make(chan struct{}), make(chan struct{}) |
| 450 | var releaseOnce sync.Once |
| 451 | release := func() { releaseOnce.Do(func() { close(proceed) }) } |
| 452 | t.Cleanup(release) |
| 453 | app.desktopSessions.beforeMigrationRegistryCommit = func() error { |
| 454 | close(entered) |
| 455 | <-proceed |
| 456 | return nil |
| 457 | } |
| 458 | interactive, err := app.prepareHistoricalSession(id, true, false) |
| 459 | if err != nil { |
| 460 | t.Fatal(err) |
| 461 | } |
| 462 | batch, err := app.prepareHistoricalSession(id, false, true) |
| 463 | if err != nil || batch != interactive { |
| 464 | t.Fatalf("interactive and batch demands did not join: %p %p %v", interactive, batch, err) |
| 465 | } |
| 466 | select { |
| 467 | case <-entered: |
| 468 | case <-time.After(10 * time.Second): |
| 469 | t.Fatal("shared preparation did not reach publication") |
| 470 | } |
| 471 | if _, err = app.CancelSessionPreparation(interactive.operationID); err != nil { |
| 472 | t.Fatal(err) |
| 473 | } |
| 474 | if err = interactive.ctx.Err(); err != nil { |
| 475 | t.Fatalf("interactive cancellation stopped the batch demand: %v", err) |
| 476 | } |
| 477 | release() |
| 478 | if result, err := waitHistoricalImport(interactive); err != nil || result.Session.SessionID == "" { |
| 479 | t.Fatalf("batch demand did not finish: %+v %v", result, err) |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | func TestHistoricalImportDoesNotReviveArchivedOrDeletedTarget(t *testing.T) { |
| 484 | for _, lifecycle := range []string{workspacestate.Archived, workspacestate.Deleted} { |
| 485 | t.Run(lifecycle, func(t *testing.T) { |
| 486 | isolateDesktopUserDirs(t) |
| 487 | root := config.SessionStoreDir() |
| 488 | coldV4MigrationFixture(t, root, "retained-source") |
| 489 | original := startupHistorySourceBytes(t, root, "retained-source") |
| 490 | app := newHistoricalLifecycleApp(t) |
| 491 | id := historicalLifecycleID(t, app, "retained-source") |
| 492 | result, err := app.ImportHistoricalSession(id) |
| 493 | if err != nil { |
| 494 | t.Fatal(err) |
| 495 | } |
| 496 | if err := app.ArchiveCanonicalSession(result.Session); err != nil { |
| 497 | t.Fatal(err) |
| 498 | } |
| 499 | if lifecycle == workspacestate.Deleted { |
| 500 | if err := app.PurgeCanonicalSession(result.Session); err != nil { |
| 501 | t.Fatal(err) |
| 502 | } |
| 503 | } |
| 504 | retired, err := app.workspaceRegistry().Load(t.Context()) |
| 505 | if err != nil { |
| 506 | t.Fatal(err) |
| 507 | } |
| 508 | retainedMemberships := len(retired.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs) |
| 509 | app.closeSessionServices() |
| 510 | app = newHistoricalLifecycleApp(t) |
| 511 | if _, err := app.ImportHistoricalSession(id); err == nil { |
| 512 | t.Fatal("explicit import silently revived a retired target") |
| 513 | } |
| 514 | if _, err := app.StartHistoricalImport(nil); err != nil { |
| 515 | t.Fatal(err) |
| 516 | } |
| 517 | awaitHistoricalBatch(t, app) |
| 518 | state, err := app.workspaceRegistry().Load(t.Context()) |
| 519 | if err != nil { |
| 520 | t.Fatal(err) |
| 521 | } |
| 522 | if state.SessionStates[result.Session.SessionID].Lifecycle != lifecycle || len(state.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs) != retainedMemberships || len(state.SourceMappings) != 1 { |
| 523 | t.Fatalf("rescan/import revived target: %+v", state) |
| 524 | } |
| 525 | if lifecycle == workspacestate.Archived { |
| 526 | assertStartupHistorySourceUnchanged(t, root, "retained-source", original) |
| 527 | } else if _, err := os.Stat(filepath.Join(root, "retained-source")); !os.IsNotExist(err) { |
| 528 | t.Fatalf("exclusive original survived purge: %v", err) |
| 529 | } |
| 530 | }) |
| 531 | } |
| 532 | } |
| 533 | |
| 534 | func TestHistoricalImportResumesPriorDurablePhase(t *testing.T) { |
| 535 | for _, scenario := range []string{"prepared", "content_ready", "content_ready_old_metadata", "prepared_old_identity", "content_ready_old_identity"} { |
| 536 | t.Run(scenario, func(t *testing.T) { |
| 537 | phase := strings.TrimSuffix(scenario, "_old_identity") |
| 538 | isolateDesktopUserDirs(t) |
| 539 | root := config.SessionStoreDir() |
| 540 | const sessionID = "interrupted-source" |
| 541 | old := coldV4MigrationFixture(t, root, sessionID) |
| 542 | app := newHistoricalLifecycleApp(t) |
| 543 | workspace, err := app.ensureDesktopWorkspace(t.Context(), "global", "") |
| 544 | if err != nil { |
| 545 | t.Fatal(err) |
| 546 | } |
| 547 | path := filepath.Join(root, sessionID) |
| 548 | fingerprint, err := desktopSourceFingerprint(path) |
| 549 | if err != nil { |
| 550 | t.Fatal(err) |
| 551 | } |
| 552 | opID, err := app.prepareDesktopImport(t.Context(), desktopMigrationSource{scope: "global"}, path, fingerprint, sessionID, workspace) |
| 553 | if err != nil { |
| 554 | t.Fatal(err) |
| 555 | } |
| 556 | if phase != "prepared" { |
| 557 | bundle := filepath.Join(t.TempDir(), "bundle") |
| 558 | if err := old.Export(t.Context(), session.SessionRef{HostID: "migration-source", SessionID: sessionID}, bundle); err != nil { |
| 559 | t.Fatal(err) |
| 560 | } |
| 561 | if _, err := app.desktopSessionService("").ImportWithHeader(t.Context(), bundle, session.CreateOptions{SessionID: sessionID, CWD: globalWorkspaceRoot(), Origin: session.SessionOriginCanonicalImport}); err != nil { |
| 562 | t.Fatal(err) |
| 563 | } |
| 564 | if phase == "content_ready_old_metadata" { |
| 565 | // Existing durable operations can omit optional presentation and |
| 566 | // retained-artifact metadata; recovery must honor that snapshot. |
| 567 | state, err := app.workspaceRegistry().Load(t.Context()) |
| 568 | if err != nil { |
| 569 | t.Fatal(err) |
| 570 | } |
| 571 | if err := app.workspaceRegistry().PrepareOperationContent(t.Context(), opID, []string{sessionID}, state.PendingOperations[opID].Mapping, nil); err != nil { |
| 572 | t.Fatal(err) |
| 573 | } |
| 574 | } else { |
| 575 | // Stop the real publication path immediately before CommitOperation, |
| 576 | // retaining its complete provenance and presentation snapshot. |
| 577 | source := desktopMigrationSource{scope: "global", operationID: opID, deferArchive: true} |
| 578 | if err := app.commitDesktopImport(t.Context(), source, path, "canonical", fingerprint, sessionID, workspace); err != nil { |
| 579 | t.Fatal(err) |
| 580 | } |
| 581 | } |
| 582 | } |
| 583 | if strings.HasSuffix(scenario, "_old_identity") { |
| 584 | rewriteHistoricalSourceIdentityForTest(t, app, desktopSourceKey(path, "")) |
| 585 | } |
| 586 | before, err := app.workspaceRegistry().Load(t.Context()) |
| 587 | if err != nil { |
| 588 | t.Fatal(err) |
| 589 | } |
| 590 | app.closeSessionServices() |
| 591 | app = newHistoricalLifecycleApp(t) |
| 592 | id := historicalLifecycleID(t, app, sessionID) |
| 593 | result, err := app.ImportHistoricalSession(id) |
| 594 | if err != nil || result.Session.SessionID != sessionID { |
| 595 | t.Fatalf("durable %s import changed identity or failed: %+v %v", phase, result, err) |
| 596 | } |
| 597 | state, err := app.workspaceRegistry().Load(t.Context()) |
| 598 | if err != nil { |
| 599 | t.Fatal(err) |
| 600 | } |
| 601 | if state.PendingOperations[opID].Phase != "committed" || len(state.SourceMappings) != 1 { |
| 602 | t.Fatalf("prior operation was not committed exactly once: %+v", state.PendingOperations) |
| 603 | } |
| 604 | if phase != "prepared" { |
| 605 | previous, committed := before.PendingOperations[opID], state.PendingOperations[opID] |
| 606 | if !reflect.DeepEqual(previous.Mapping, committed.Mapping) || !reflect.DeepEqual(previous.Presentation, committed.Presentation) { |
| 607 | t.Fatal("resuming content_ready rewrote its durable metadata") |
| 608 | } |
| 609 | } |
| 610 | if _, err := app.desktopSessionService("").Query().Snapshot(t.Context(), result.Session); err != nil { |
| 611 | t.Fatal(err) |
| 612 | } |
| 613 | }) |
| 614 | } |
| 615 | } |
| 616 |