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