From 05de0e8ebf48875f234d23f379a52d1749f965ab Mon Sep 17 00:00:00 2001 From: zy C Date: Tue, 27 Oct 2026 06:48:58 -0700 Subject: [PATCH 1/3] Recover stale timeline date continuations --- app/journal_graph_feed_plan.ml | 20 +++- test/journal_graph_runtime_locality_test.ml | 102 ++++++++++++++++++++ 2 files changed, 121 insertions(+), 1 deletion(-) diff --git a/app/journal_graph_feed_plan.ml b/app/journal_graph_feed_plan.ml index 46c5897..87e5e38 100644 --- a/app/journal_graph_feed_plan.ml +++ b/app/journal_graph_feed_plan.ml @@ -13,7 +13,7 @@ type feed_scan = { root_session : int64 ; root_generation : int64 ; root_start_day : int option - ; root_before_day : int option + ; mutable root_before_day : int option ; root_skipped_day : int option ; mutable root_date_boundary : int option ; mutable root_priority_pending : bool @@ -25,6 +25,8 @@ type feed_scan = ; mutable root_days : day_feed Rrbvec.t ; mutable root_queued : Projection.page Rrbvec.t ; mutable root_journal_cursor : Graph.Cursor.t option + ; mutable root_journal_boundary : int option + ; mutable root_date_retries : int ; mutable root_journals_done : bool } @@ -735,6 +737,8 @@ let start_feed_scan t ~ports ~cursor ~root_limit ~day_limit ~request_generation ; root_days = Rrbvec.empty ; root_queued = Rrbvec.empty ; root_journal_cursor = None + ; root_journal_boundary = before_day + ; root_date_retries = 0 ; root_journals_done = after <> None && cursor = None } in @@ -777,6 +781,8 @@ let handles operation outcome = match operation,outcome with | Root_tree _, Protocol.V2_page_tree_outcome _ | Root_preview _, Protocol.V2_children_outcome _ -> true | Root_dates scan, Protocol.V2_failed {code="responseTooLarge";_} -> scan.root_list_size > 1 + | Root_dates scan, Protocol.V2_failed {code="staleReadCursor";_} -> + Option.is_some scan.root_journal_cursor && scan.root_date_retries < 2 | Root_tree batch, Protocol.V2_failed {code="responseTooLarge";_} -> batch.root_limit > 1 | Root_preview preview, Protocol.V2_failed {code="responseTooLarge";_} -> preview.preview_limit > 1 | _ -> false @@ -803,6 +809,12 @@ let receive t ~ports operation outcome = pages in scan.root_queued <- pages; + (* Catalog progress includes Today, even when the feed continuation skips it. *) + scan.root_journal_boundary <- + Rrbvec.fold_left + (fun boundary (item : Protocol.v2_journal_item) -> + Some (Option.fold ~none:item.journal_day ~some:(min item.journal_day) boundary)) + scan.root_journal_boundary items; scan.root_journal_cursor <- next_cursor; scan.root_journals_done <- next_cursor = None; if scan.root_priority_pending @@ -844,6 +856,12 @@ let receive t ~ports operation outcome = | Root_preview preview, V2_children_outcome {parent;scope_revision;items;next_cursor;_} -> ports.observe_packet (Preview (List.to_seq items)); receive_root_preview t ~ports preview parent scope_revision items next_cursor + | Root_dates scan, V2_failed {code="staleReadCursor";_} -> + if not (root_owner_current t (Feed_roots scan)) then empty else ( + scan.root_date_retries <- scan.root_date_retries + 1; + scan.root_before_day <- scan.root_journal_boundary; + scan.root_journal_cursor <- None; + requests [read_root_dates ~ports scan]) | Root_dates scan, V2_failed {code="responseTooLarge";_} when scan.root_list_size>1 -> scan.root_list_size <- max 1 (scan.root_list_size/2);requests [read_root_dates ~ports scan] | Root_tree batch, V2_failed {code="responseTooLarge";_} when batch.root_limit>1 -> diff --git a/test/journal_graph_runtime_locality_test.ml b/test/journal_graph_runtime_locality_test.ml index 9a50ffd..3ed220c 100644 --- a/test/journal_graph_runtime_locality_test.ml +++ b/test/journal_graph_runtime_locality_test.ml @@ -7691,6 +7691,107 @@ let test_later_thirty_crosses_dates () = Alcotest.(check int) "continuations do not repeat roots" 60 (List.length (List.sort_uniq String.compare all)) ;; +let test_date_cursor_recovery_is_bounded_and_owned () = + (* Worker failures exercise recovery, not storage cursor invalidation. *) + List.iter + (fun ending -> + let runtime = Production_runtime.create ~localtime:Unix.gmtime () in + set_calendar runtime; + let first = normal_journal_fixture 29 1 0 in + let older = normal_journal_fixture 30 1 0 in + let listing = + Production_runtime.submit runtime (Journal_graph_request.Load_feed + { cursor = None; root_limit = 3; day_limit = 1; request_generation = 77L }) + |> fun output -> only "initial date catalog" output.requests + in + let complete_date listing fixture next_cursor expected_limit = + let journal_day = + match fixture.owner.kind with + | Journal_page { journal_day } -> journal_day + | _ -> assert false + in + let tree = + Production_runtime.receive runtime (response listing (V2_journals_outcome + { items = [ { page = fixture.owner; journal_day; revision = "page-1" } ] + ; next_cursor })) + |> fun output -> only "date root read" output.requests + in + (match tree.command with + | V2_get_page_tree { limit; _ } -> + Alcotest.(check int) "retains remaining root quota" expected_limit limit + | _ -> Alcotest.fail "expected root query"); + let root = only "fixture root" fixture.roots in + let preview = + Production_runtime.receive runtime (response tree (V2_page_tree_outcome + { page = fixture.owner.uuid; maximum_depth = 0 + ; items = [ { value = root; revision = "root-1"; depth = 0 + ; parent = fixture.owner.uuid } ] + ; next_cursor = None })) + |> fun output -> only "root preview" output.requests + in + Production_runtime.receive runtime (response preview (V2_children_outcome + { parent = root.block.uuid + ; revision_scope = V2_children_revision root.block.uuid + ; scope_revision = "children-1"; items = []; next_cursor = None })) + in + let stale request = + Production_runtime.receive runtime (response request (V2_failed + { code = "staleReadCursor"; message = "The read continuation is no longer current." })) + in + let recover request through_day = + let output = stale request in + Alcotest.(check int) "recoverable stale read is not published" 0 + (List.length output.responses); + let retry = only "bounded date recovery" output.requests in + (match retry.command with + | V2_list_journals { cursor; through_day = actual; limit; _ } -> + Alcotest.(check bool) "discard obsolete worker cursor" true (cursor = None); + Alcotest.(check int) "resume strictly after consumed dates" through_day actual; + Alcotest.(check int) "retain bounded metadata packet" 1 limit + | _ -> Alcotest.fail "expected date recovery"); + let duplicate = stale request in + Alcotest.(check int) "duplicate failure cannot retry" 0 + (List.length duplicate.requests + List.length duplicate.responses); + retry + in + let output = complete_date listing first (Some (root_cursor 1)) 3 in + let retry = recover (only "first date continuation" output.requests) 20260831 in + let output = complete_date retry older (Some (root_cursor 2)) 2 in + let retry = recover (only "second date continuation" output.requests) 20260830 in + match ending with + | `Complete -> + let output = Production_runtime.receive runtime + (response retry (V2_journals_outcome { items = []; next_cursor = None })) in + (match (only "recovered feed completion" output.responses).payload with + | Feed_loaded { request_generation = 77L; feed; complete = true } -> + Alcotest.(check (list string)) "preserve prefix without duplicate roots" + (List.map (fun f -> Graph.Uuid.to_string (only "fixture root" f.roots).block.uuid) + [ first; older ]) + (List.concat_map (fun (day : Journal_graph_projection.day_feed) -> + List.map (fun (entry : Journal_graph_projection.timeline_entry) -> + Journal_model.id entry.block) day.entries) feed.days) + | _ -> Alcotest.fail "date recovery lost feed ownership") + | `Exhaust -> + let output = Production_runtime.receive runtime + (response retry (V2_journals_outcome + { items = []; next_cursor = Some (root_cursor 3) })) in + let output = stale (only "third date continuation" output.requests) in + Alcotest.(check int) "finite recovery budget" 0 (List.length output.requests); + (match (only "terminal stale cursor failure" output.responses).payload with + | Feed_failed { request_generation = 77L; failure = Worker_failure failure } -> + Alcotest.(check string) "terminal operation retained" "listTimelineDates" + failure.operation + | _ -> Alcotest.fail "retry exhaustion lost failure ownership") + | `Supersede -> + ignore (Production_runtime.submit runtime (Journal_graph_request.Load_feed + { cursor = None; root_limit = 3; day_limit = 1; request_generation = 78L })); + let output = Production_runtime.receive runtime + (response retry (V2_journals_outcome { items = []; next_cursor = None })) in + Alcotest.(check int) "superseded recovery cannot publish or continue" 0 + (List.length output.requests + List.length output.responses)) + [ `Complete; `Exhaust; `Supersede ] +;; + let test_later_date_cursor_no_progress_settles () = let runtime=Production_runtime.create ~localtime:Unix.gmtime () in set_calendar runtime; @@ -8384,6 +8485,7 @@ let () = Alcotest.test_case "restart retains loaded suffix through insertion" `Quick test_initial_refresh_after_restarting_prefix; Alcotest.test_case "superseded continuation settles" `Quick test_initial_later_read_superseded; Alcotest.test_case "later batch fills across dates" `Quick test_later_thirty_crosses_dates; + Alcotest.test_case "date cursor recovery is bounded and owned" `Quick test_date_cursor_recovery_is_bounded_and_owned; Alcotest.test_case "later date cursor failure settles" `Quick test_later_date_cursor_no_progress_settles; Alcotest.test_case "resync prefix cannot cancel retained tree" `Quick (test_prefix_preserves_reconciliation ~resync:true `Root); From f20b3ce2a6bd0059c307f22f8ff206eb62dede31 Mon Sep 17 00:00:00 2001 From: zy C Date: Tue, 27 Oct 2026 06:57:21 -0700 Subject: [PATCH 2/3] Recover stale dates without skipping same-day pages --- app/journal_graph_feed_plan.ml | 40 +++++-- test/journal_graph_runtime_locality_test.ml | 116 ++++++++++++-------- 2 files changed, 100 insertions(+), 56 deletions(-) diff --git a/app/journal_graph_feed_plan.ml b/app/journal_graph_feed_plan.ml index 87e5e38..ed9e207 100644 --- a/app/journal_graph_feed_plan.ml +++ b/app/journal_graph_feed_plan.ml @@ -13,7 +13,7 @@ type feed_scan = { root_session : int64 ; root_generation : int64 ; root_start_day : int option - ; mutable root_before_day : int option + ; root_before_day : int option ; root_skipped_day : int option ; mutable root_date_boundary : int option ; mutable root_priority_pending : bool @@ -25,7 +25,9 @@ type feed_scan = ; mutable root_days : day_feed Rrbvec.t ; mutable root_queued : Projection.page Rrbvec.t ; mutable root_journal_cursor : Graph.Cursor.t option + ; mutable root_journal_through : int ; mutable root_journal_boundary : int option + ; root_journal_boundary_pages : (string, unit) Hashtbl.t ; mutable root_date_retries : int ; mutable root_journals_done : bool } @@ -259,8 +261,7 @@ let read_root_dates ~ports scan = (Root_dates scan) (Protocol.V2_list_journals { from_day = 0 - ; through_day = - Option.fold ~none:99_999_999 ~some:previous_journal_day scan.root_before_day + ; through_day = scan.root_journal_through ; limit = scan.root_list_size ; cursor = scan.root_journal_cursor ; revision = None @@ -737,7 +738,10 @@ let start_feed_scan t ~ports ~cursor ~root_limit ~day_limit ~request_generation ; root_days = Rrbvec.empty ; root_queued = Rrbvec.empty ; root_journal_cursor = None - ; root_journal_boundary = before_day + ; root_journal_through = + Option.fold ~none:99_999_999 ~some:previous_journal_day before_day + ; root_journal_boundary = None + ; root_journal_boundary_pages = Hashtbl.create 8 ; root_date_retries = 0 ; root_journals_done = after <> None && cursor = None } @@ -790,7 +794,12 @@ let receive t ~ports operation outcome = if not (handles operation outcome) then None else Some (match operation,outcome with | Root_dates scan, Protocol.V2_journals_outcome { items; next_cursor } -> ports.observe_packet (Dates (List.to_seq items)); - let items = Rrbvec.of_list items in + let boundary_page (i : Protocol.v2_journal_item) = + scan.root_journal_boundary = Some i.journal_day + && Hashtbl.mem scan.root_journal_boundary_pages (Graph.Uuid.to_string i.page.uuid) + in + (* Stale recovery rereads the oldest catalog date inclusively. *) + let items = Rrbvec.of_list items |> Rrbvec.filter (fun i -> not (boundary_page i)) in if next_cursor <> None && next_cursor = scan.root_journal_cursor then scan_failure scan "The date cursor did not advance." else ( @@ -810,11 +819,20 @@ let receive t ~ports operation outcome = in scan.root_queued <- pages; (* Catalog progress includes Today, even when the feed continuation skips it. *) - scan.root_journal_boundary <- - Rrbvec.fold_left - (fun boundary (item : Protocol.v2_journal_item) -> - Some (Option.fold ~none:item.journal_day ~some:(min item.journal_day) boundary)) - scan.root_journal_boundary items; + Rrbvec.iter + (fun (i : Protocol.v2_journal_item) -> + if Option.fold ~none:true ~some:(fun day -> i.journal_day < day) + scan.root_journal_boundary + then ( + scan.root_journal_boundary <- Some i.journal_day; + Hashtbl.reset scan.root_journal_boundary_pages); + if scan.root_journal_boundary = Some i.journal_day + then + Hashtbl.replace + scan.root_journal_boundary_pages + (Graph.Uuid.to_string i.page.uuid) + ()) + items; scan.root_journal_cursor <- next_cursor; scan.root_journals_done <- next_cursor = None; if scan.root_priority_pending @@ -859,7 +877,7 @@ let receive t ~ports operation outcome = | Root_dates scan, V2_failed {code="staleReadCursor";_} -> if not (root_owner_current t (Feed_roots scan)) then empty else ( scan.root_date_retries <- scan.root_date_retries + 1; - scan.root_before_day <- scan.root_journal_boundary; + Option.iter (fun day -> scan.root_journal_through <- day) scan.root_journal_boundary; scan.root_journal_cursor <- None; requests [read_root_dates ~ports scan]) | Root_dates scan, V2_failed {code="responseTooLarge";_} when scan.root_list_size>1 -> diff --git a/test/journal_graph_runtime_locality_test.ml b/test/journal_graph_runtime_locality_test.ml index 3ed220c..00f83c6 100644 --- a/test/journal_graph_runtime_locality_test.ml +++ b/test/journal_graph_runtime_locality_test.ml @@ -7696,26 +7696,54 @@ let test_date_cursor_recovery_is_bounded_and_owned () = List.iter (fun ending -> let runtime = Production_runtime.create ~localtime:Unix.gmtime () in - set_calendar runtime; + set_calendar ~clock:1_788_307_200. runtime; let first = normal_journal_fixture 29 1 0 in let older = normal_journal_fixture 30 1 0 in - let listing = - Production_runtime.submit runtime (Journal_graph_request.Load_feed - { cursor = None; root_limit = 3; day_limit = 1; request_generation = 77L }) - |> fun output -> only "initial date catalog" output.requests + let same_day = + let id = uuid "00000002-2026-0901-0000-000000000000" in + let root = only "fixture root" first.roots in + { first with + owner = { first.owner with uuid = id } + ; roots = + [ { root with + block = + { root.block with + uuid = uuid "00000002-2026-0901-0000-000000000001" + ; parent = id + ; page = id + } + } ] + ; descendants = [] + } in - let complete_date listing fixture next_cursor expected_limit = - let journal_day = - match fixture.owner.kind with - | Journal_page { journal_day } -> journal_day - | _ -> assert false - in - let tree = - Production_runtime.receive runtime (response listing (V2_journals_outcome - { items = [ { page = fixture.owner; journal_day; revision = "page-1" } ] - ; next_cursor })) - |> fun output -> only "date root read" output.requests - in + let day fixture = + match fixture.owner.kind with + | Journal_page { journal_day } -> journal_day + | _ -> assert false + in + let dates request fixtures next_cursor = + Production_runtime.receive runtime (response request (V2_journals_outcome + { items = List.map (fun f -> ({ page = f.owner; journal_day = day f + ; revision = "page-1" } : Protocol.v2_journal_item)) + fixtures + ; next_cursor })) + in + let listing ?(quiet = true) (output : Production_runtime.output) cursor through = + if quiet then + Alcotest.(check int) "date recovery publishes nothing" 0 + (List.length output.responses); + let request = only "one date catalog read" output.requests in + (match request.command with + | V2_list_journals { cursor = actual; through_day; limit; _ } -> + Alcotest.(check bool) "date continuation" true (actual = cursor); + Alcotest.(check int) "date query bound" through through_day; + Alcotest.(check int) "bounded metadata packet" 1 limit + | _ -> Alcotest.fail "expected only a date catalog read"); + request + in + let complete_date listing fixture next_cursor expected_limit through = + let tree = dates listing [ fixture ] next_cursor + |> fun output -> only "date root read" output.requests in (match tree.command with | V2_get_page_tree { limit; _ } -> Alcotest.(check int) "retains remaining root quota" expected_limit limit @@ -7729,53 +7757,52 @@ let test_date_cursor_recovery_is_bounded_and_owned () = ; next_cursor = None })) |> fun output -> only "root preview" output.requests in - Production_runtime.receive runtime (response preview (V2_children_outcome + let output = Production_runtime.receive runtime (response preview (V2_children_outcome { parent = root.block.uuid ; revision_scope = V2_children_revision root.block.uuid - ; scope_revision = "children-1"; items = []; next_cursor = None })) + ; scope_revision = "children-1"; items = []; next_cursor = None })) in + listing ~quiet:false output next_cursor through in let stale request = Production_runtime.receive runtime (response request (V2_failed { code = "staleReadCursor"; message = "The read continuation is no longer current." })) in - let recover request through_day = - let output = stale request in - Alcotest.(check int) "recoverable stale read is not published" 0 - (List.length output.responses); - let retry = only "bounded date recovery" output.requests in - (match retry.command with - | V2_list_journals { cursor; through_day = actual; limit; _ } -> - Alcotest.(check bool) "discard obsolete worker cursor" true (cursor = None); - Alcotest.(check int) "resume strictly after consumed dates" through_day actual; - Alcotest.(check int) "retain bounded metadata packet" 1 limit - | _ -> Alcotest.fail "expected date recovery"); + (* Recovery rereads the oldest observed date inclusively. *) + let recover request = + let retry = listing (stale request) None 20260901 in let duplicate = stale request in Alcotest.(check int) "duplicate failure cannot retry" 0 (List.length duplicate.requests + List.length duplicate.responses); retry in - let output = complete_date listing first (Some (root_cursor 1)) 3 in - let retry = recover (only "first date continuation" output.requests) 20260831 in - let output = complete_date retry older (Some (root_cursor 2)) 2 in - let retry = recover (only "second date continuation" output.requests) 20260830 in + let replay request fixtures cursor = + listing (dates request fixtures (Some cursor)) (Some cursor) 20260901 + in + let initial = + Production_runtime.submit runtime (Journal_graph_request.Load_feed + { cursor = None; root_limit = 4; day_limit = 1; request_generation = 77L }) + |> fun output -> listing ~quiet:false output None 99_999_999 + in + let next = complete_date initial first (Some (root_cursor 1)) 4 99_999_999 in + let next = replay (recover next) [ first ] (root_cursor 2) in + let next = complete_date next same_day (Some (root_cursor 3)) 3 20260901 in + let next = replay (recover next) [ first ] (root_cursor 4) in + let next = replay next [ same_day ] (root_cursor 5) in + let next = complete_date next older (Some (root_cursor 6)) 2 20260901 in match ending with | `Complete -> - let output = Production_runtime.receive runtime - (response retry (V2_journals_outcome { items = []; next_cursor = None })) in + let output = dates next [] None in (match (only "recovered feed completion" output.responses).payload with | Feed_loaded { request_generation = 77L; feed; complete = true } -> - Alcotest.(check (list string)) "preserve prefix without duplicate roots" + Alcotest.(check (list string)) "same-day pages once, consumed roots once" (List.map (fun f -> Graph.Uuid.to_string (only "fixture root" f.roots).block.uuid) - [ first; older ]) + [ first; same_day; older ]) (List.concat_map (fun (day : Journal_graph_projection.day_feed) -> List.map (fun (entry : Journal_graph_projection.timeline_entry) -> Journal_model.id entry.block) day.entries) feed.days) | _ -> Alcotest.fail "date recovery lost feed ownership") | `Exhaust -> - let output = Production_runtime.receive runtime - (response retry (V2_journals_outcome - { items = []; next_cursor = Some (root_cursor 3) })) in - let output = stale (only "third date continuation" output.requests) in + let output = stale next in Alcotest.(check int) "finite recovery budget" 0 (List.length output.requests); (match (only "terminal stale cursor failure" output.responses).payload with | Feed_failed { request_generation = 77L; failure = Worker_failure failure } -> @@ -7784,9 +7811,8 @@ let test_date_cursor_recovery_is_bounded_and_owned () = | _ -> Alcotest.fail "retry exhaustion lost failure ownership") | `Supersede -> ignore (Production_runtime.submit runtime (Journal_graph_request.Load_feed - { cursor = None; root_limit = 3; day_limit = 1; request_generation = 78L })); - let output = Production_runtime.receive runtime - (response retry (V2_journals_outcome { items = []; next_cursor = None })) in + { cursor = None; root_limit = 4; day_limit = 1; request_generation = 78L })); + let output = dates next [] None in Alcotest.(check int) "superseded recovery cannot publish or continue" 0 (List.length output.requests + List.length output.responses)) [ `Complete; `Exhaust; `Supersede ] From ffd787c839aa9ae9645d0f2aada097b94fc1d8b6 Mon Sep 17 00:00:00 2001 From: zy C Date: Tue, 27 Oct 2026 06:58:51 -0700 Subject: [PATCH 3/3] Correct recovery regression fixture compilation --- test/journal_graph_runtime_locality_test.ml | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/test/journal_graph_runtime_locality_test.ml b/test/journal_graph_runtime_locality_test.ml index 00f83c6..4626fb7 100644 --- a/test/journal_graph_runtime_locality_test.ml +++ b/test/journal_graph_runtime_locality_test.ml @@ -7702,8 +7702,7 @@ let test_date_cursor_recovery_is_bounded_and_owned () = let same_day = let id = uuid "00000002-2026-0901-0000-000000000000" in let root = only "fixture root" first.roots in - { first with - owner = { first.owner with uuid = id } + { owner = { first.owner with uuid = id } ; roots = [ { root with block = @@ -7741,8 +7740,8 @@ let test_date_cursor_recovery_is_bounded_and_owned () = | _ -> Alcotest.fail "expected only a date catalog read"); request in - let complete_date listing fixture next_cursor expected_limit through = - let tree = dates listing [ fixture ] next_cursor + let complete_date request fixture next_cursor expected_limit through = + let tree = dates request [ fixture ] next_cursor |> fun output -> only "date root read" output.requests in (match tree.command with | V2_get_page_tree { limit; _ } ->