| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "testing" |
| 6 | "time" |
| 7 | |
| 8 | "reasonix/internal/projectiondb" |
| 9 | ) |
| 10 | |
| 11 | func TestObsoleteHistoryPreparesWithoutBlockingAndRefreshesAfterAppend(t *testing.T) { |
| 12 | s, r, _, path := historyBoundaryFixture(t) |
| 13 | q := s.Query() |
| 14 | _ = historyPageReady(t, q, r.Ref(), "", 100) |
| 15 | h, err := projectiondb.Open(t.Context(), projectiondb.OpenOptions{Path: path, Migrations: historyMigrations, RequireDisk: true}) |
| 16 | if err != nil { |
| 17 | t.Fatal(err) |
| 18 | } |
| 19 | _, err = h.DB.ExecContext(t.Context(), "UPDATE metadata SET value='0' WHERE key='projection_version'") |
| 20 | _ = h.DB.Close() |
| 21 | if err != nil { |
| 22 | t.Fatal(err) |
| 23 | } |
| 24 | // Hold both slots: an obsolete on-disk index must return preparing, not |
| 25 | // rebuild synchronously or wait until another UI action frees capacity. |
| 26 | q.slots.tryAcquire() |
| 27 | q.slots.tryAcquire() |
| 28 | released := false |
| 29 | defer func() { |
| 30 | if !released { |
| 31 | q.slots.release() |
| 32 | q.slots.release() |
| 33 | } |
| 34 | }() |
| 35 | assertHistoryPreparing(t, q, r.Ref()) |
| 36 | q.slots.release() |
| 37 | q.slots.release() |
| 38 | released = true |
| 39 | page := windowReady(t, q, r.Ref(), HistoryWindowRequest{Anchor: "newest"}) |
| 40 | if len(page.Messages) != 1 { |
| 41 | t.Fatalf("recovered messages=%d", len(page.Messages)) |
| 42 | } |
| 43 | // The completed preparation handle must not conceal later invalidation. |
| 44 | appendRecoveryTestMessage(t, r.Session(), "second", "second") |
| 45 | if _, err := r.Session().Flush(t.Context()); err != nil { |
| 46 | t.Fatal(err) |
| 47 | } |
| 48 | page = windowReady(t, q, r.Ref(), HistoryWindowRequest{Anchor: "newest"}) |
| 49 | if len(page.Messages) != 2 { |
| 50 | t.Fatalf("appended messages=%d", len(page.Messages)) |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | func TestHistoryReadersDoNotWaitOnRebuildMutex(t *testing.T) { |
| 55 | s, r, _, _ := historyBoundaryFixture(t) |
| 56 | q := s.Query() |
| 57 | _ = historyPageReady(t, q, r.Ref(), "", 100) |
| 58 | lock := q.projectionLock("history", r.Ref().SessionID) |
| 59 | lock.Lock() |
| 60 | defer lock.Unlock() |
| 61 | assertHistoryPreparing(t, q, r.Ref()) |
| 62 | } |
| 63 | |
| 64 | func assertHistoryPreparing(t *testing.T, q *Query, ref SessionRef) { |
| 65 | t.Helper() |
| 66 | readers := []func(context.Context) (string, error){ |
| 67 | func(ctx context.Context) (string, error) { |
| 68 | p, e := q.ReadHistoryWindow(ctx, ref, HistoryWindowRequest{Anchor: "newest"}) |
| 69 | return p.Status, e |
| 70 | }, |
| 71 | func(ctx context.Context) (string, error) { |
| 72 | p, e := q.HistoryPage(ctx, ref, "", 100) |
| 73 | return p.Status, e |
| 74 | }, |
| 75 | func(ctx context.Context) (string, error) { |
| 76 | p, e := q.LocateMessage(ctx, ref, "first", 0) |
| 77 | return p.Status, e |
| 78 | }, |
| 79 | } |
| 80 | for i, read := range readers { |
| 81 | ctx, cancel := context.WithTimeout(t.Context(), time.Second) |
| 82 | type result struct { |
| 83 | status string |
| 84 | err error |
| 85 | } |
| 86 | done := make(chan result, 1) |
| 87 | go func() { status, err := read(ctx); done <- result{status, err} }() |
| 88 | select { |
| 89 | case p := <-done: |
| 90 | cancel() |
| 91 | if p.err != nil || p.status != "preparing" { |
| 92 | t.Fatalf("reader %d: %+v", i, p) |
| 93 | } |
| 94 | case <-ctx.Done(): |
| 95 | cancel() |
| 96 | t.Fatalf("reader %d blocked behind rebuild", i) |
| 97 | } |
| 98 | } |
| 99 | } |
| 100 |