| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "reasonix/internal/config" |
| 11 | "reasonix/internal/sessioncatalog" |
| 12 | ) |
| 13 | |
| 14 | func TestConcurrentRebuildSessionCatalogCallersShareFailure(t *testing.T) { |
| 15 | isolateDesktopUserDirs(t) |
| 16 | dir := config.SessionDir() |
| 17 | if err := os.MkdirAll(dir, 0o755); err != nil { |
| 18 | t.Fatal(err) |
| 19 | } |
| 20 | |
| 21 | app := NewApp() |
| 22 | app.startSessionCatalog() |
| 23 | _ = waitForSessionCatalogForTest(t, app, nil) |
| 24 | t.Cleanup(func() { app.stopSessionCatalog(time.Second) }) |
| 25 | |
| 26 | reconcileStarted := make(chan struct{}) |
| 27 | reconcileRelease := make(chan struct{}) |
| 28 | var reconcileDone []<-chan struct{} |
| 29 | t.Cleanup(func() { |
| 30 | close(reconcileRelease) |
| 31 | for _, done := range reconcileDone { |
| 32 | if !waitChannelBefore(done, time.Now().Add(5*time.Second)) { |
| 33 | t.Error("reconcile did not exit after release") |
| 34 | } |
| 35 | } |
| 36 | }) |
| 37 | app.catalogReconcileHook = func(sessioncatalog.DirectoryTarget) { |
| 38 | close(reconcileStarted) |
| 39 | <-reconcileRelease |
| 40 | } |
| 41 | if !app.requestSessionCatalogReconcile(dir) { |
| 42 | t.Fatal("explicit reconcile was not scheduled") |
| 43 | } |
| 44 | <-reconcileStarted |
| 45 | app.catalogReconcileMu.Lock() |
| 46 | for _, job := range app.catalogReconcileJobs { |
| 47 | reconcileDone = append(reconcileDone, job.done) |
| 48 | } |
| 49 | app.catalogReconcileMu.Unlock() |
| 50 | |
| 51 | app.ctx = context.Background() |
| 52 | rebuildStarted := make(chan struct{}) |
| 53 | joined := make(chan struct{}) |
| 54 | rebuildRelease := make(chan struct{}, 1) |
| 55 | defer close(rebuildRelease) |
| 56 | app.runtimeEvents.emit = func(_ context.Context, name string, payload ...any) { |
| 57 | if name != "project-tree:changed-v2" || len(payload) != 1 { |
| 58 | return |
| 59 | } |
| 60 | event, ok := payload[0].(ProjectTreeChangedV2) |
| 61 | if ok && event.Reason == "catalog_rebuild_started" { |
| 62 | close(rebuildStarted) |
| 63 | // Keep the flight published until the follower has joined. Hold |
| 64 | // reconcile blocked through both results to force the exact stop |
| 65 | // failure independently of SQLite/disk speed. |
| 66 | <-rebuildRelease |
| 67 | } |
| 68 | } |
| 69 | app.catalogRebuildJoinHook = func() { close(joined) } |
| 70 | |
| 71 | leaderDone := make(chan error, 1) |
| 72 | go func() { leaderDone <- app.rebuildSessionCatalog(25 * time.Millisecond) }() |
| 73 | <-rebuildStarted |
| 74 | followerDone := make(chan error, 1) |
| 75 | go func() { followerDone <- app.rebuildSessionCatalog(25 * time.Millisecond) }() |
| 76 | <-joined |
| 77 | select { |
| 78 | case err := <-followerDone: |
| 79 | t.Fatalf("concurrent rebuild returned before the owner completed: %v", err) |
| 80 | default: |
| 81 | } |
| 82 | |
| 83 | rebuildRelease <- struct{}{} |
| 84 | leaderErr := <-leaderDone |
| 85 | followerErr := <-followerDone |
| 86 | if leaderErr == nil || followerErr == nil { |
| 87 | t.Fatalf("concurrent rebuild errors = leader %v, follower %v; want shared failure", leaderErr, followerErr) |
| 88 | } |
| 89 | if !errors.Is(followerErr, leaderErr) || !errors.Is(leaderErr, errSessionCatalogStopTimeout) { |
| 90 | t.Fatalf("concurrent rebuild errors = leader %q, follower %q; want the same rebuild result", |
| 91 | leaderErr, followerErr) |
| 92 | } |
| 93 | if app.catalogRebuilding.Load() { |
| 94 | t.Fatal("shared failed rebuild left rebuilding set") |
| 95 | } |
| 96 | } |
| 97 |