Recover stale journal date cursors without losing same-day pages

app/journal_graph_feed_plan.ml

@@ -25,6 +25,10 @@ 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
   }
 
@@ -257,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
@@ -735,6 +738,11 @@ 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_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
       }
     in
@@ -777,14 +785,21 @@ 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
 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 (
@@ -803,6 +818,21 @@ let receive t ~ports operation outcome =
                 pages
           in
           scan.root_queued <- pages;
+          (* Catalog progress includes Today, even when the feed continuation skips it. *)
+          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
@@ -844,6 +874,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;
+         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 ->
        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 ->

test/journal_graph_runtime_locality_test.ml

@@ -7691,6 +7691,132 @@ 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 ~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 same_day =
+         let id = uuid "00000002-2026-0901-0000-000000000000" in
+         let root = only "fixture root" first.roots in
+         { 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 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 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; _ } ->
+            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
+         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 })) 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
+       (* 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 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 = 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)) "same-day pages once, consumed roots once"
+              (List.map (fun f -> Graph.Uuid.to_string (only "fixture root" f.roots).block.uuid)
+                 [ 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 = 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 } ->
+            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 = 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 ]
+;;
+
 let test_later_date_cursor_no_progress_settles () =
   let runtime=Production_runtime.create ~localtime:Unix.gmtime () in
   set_calendar runtime;
@@ -8384,6 +8510,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);