Fix Capture sync wakeups, deletion replay and preparation races

logseq_db_worker/lib/effect_runner/effect_runner.ml

@@ -1951,7 +1951,12 @@ let run_durable_mutation t ~(scope:Sync.graph_scope) ~intent ~current =
          | Ok(Database.Intent_rejected error)->reject error
          | Error error->report_commit error; Mutation.Unknown
          | Ok _->Mutation.Unknown)
-      |Ok proof->Mutation.Committed(commit_proof(Database.intent_commit_outcome proof))
+      |Ok proof->
+        (* A durable commit (including replay) must wake its own graph even if
+           the caller registration retired while storage was committing. The
+           scoped sync owner rejects retired graph/account generations. *)
+        if not t.stopped then t.post(Core.Sync_event(Sync.Local_graph_outbox_changed scope));
+        Mutation.Committed(commit_proof(Database.intent_commit_outcome proof))
 ;;
 
 let durable_mutation t ~scope ~intent ~current =
@@ -2778,13 +2783,24 @@ let handle_sync_worker_effect t = function
     (match attached_session t request.scope with
      | None -> Error (effect_error "The overlay database is unavailable.")
      | Some session ->
+       let failed error =
+         match error with
+         | Overlay.Outbox_sync_token_conflict | Outbox_commit_token_conflict
+         | Outbox_crypto_error Overlay.Crypto_result_stale ->
+           (* These failures precede persistence. A newer observed revision, not
+              the error text alone, is required before asking policy to replan. *)
+           (match Database.inspect_sync session.database with
+            | Ok sync when not(Overlay.sync_token_equal request.expected(Overlay.sync_view_token sync))->
+              Ok(sync_result ~event:(Sync.Outbox_transition_superseded{request;sync})())
+            | Ok _ | Error _ -> Error(outbox_transition_error error))
+         | _ -> Error(outbox_transition_error error) in
        (match
           Database.begin_outbox_transition
             session.database
             ~expected:request.expected
             request.transition
         with
-        | Error error -> Error (outbox_transition_error error)
+        | Error error -> failed error
         | Ok (prepared, crypto) ->
           let encrypted =
             match crypto with
@@ -2797,7 +2813,7 @@ let handle_sync_worker_effect t = function
              (match
                 Database.apply_outbox_transition session.database prepared ~encrypted
               with
-              | Error error -> Error (outbox_transition_error error)
+              | Error error -> failed error
               | Ok commit ->
                 (match Database.inspect_sync session.database with
                  | Error error -> Error (sync_read_error error)

logseq_overlay_db/lib/database.ml

@@ -6761,6 +6761,7 @@ let descriptor (record : outbox_record) =
   { Types.mutation_id = record.mutation_id
   ; fingerprint = mutation_fingerprint record.fingerprint
   ; state = record.transport_state
+  ; operation = operation_of_mutation record.mutation
   ; dependency_eligible = record.transport_state <> Types.Blocked
   ; attempt_count = record.attempt_count
   ; plaintext_bytes = String.length record.normalized_transaction
@@ -10105,17 +10106,54 @@ let commit_authoritative_candidate database prepared =
             in
             let authoritative_after = prepared.authoritative_db_after in
             let authoritative_before = authoritative_database database in
+            (* Origin validation compares complete transaction data at its expected
+               cursor. Earlier local effects were already visible when a queued
+               delete was authored; their echoes are not competing remote edits. *)
+            let local_sequences = Hashtbl.create (List.length database.outbox) in
+            List.iter (fun (record : outbox_record) ->
+              Hashtbl.replace local_sequences (Graph.Uuid.to_string record.mutation_id)
+                record.sequence) database.outbox;
+            let origin_sequences = Hashtbl.create (List.length prepared.validated_origins) in
+            List.iter (fun (id, cursor) ->
+              Option.iter (fun sequence ->
+                Hashtbl.replace origin_sequences (Types.Server_cursor.to_int64 cursor) sequence)
+                (Hashtbl.find_opt local_sequences (Graph.Uuid.to_string id)))
+              prepared.validated_origins;
+            let _, conflict_contexts = List.fold_left2
+              (fun (before, contexts) transaction after ->
+                let origin = Hashtbl.find_opt origin_sequences
+                  (Types.Server_cursor.to_int64
+                    (Types.authoritative_transaction_cursor transaction)) in
+                after, Rrbvec.push_back contexts (before, after, origin))
+              (authoritative_before, Rrbvec.empty)
+              (Types.authoritative_batch_transactions preparation.authoritative_batch)
+              prepared.authoritative_roots_after in
+            let module Conflict_set = Set.Make (struct
+              type t = Types.delete_conflict_kind
+              let compare = Stdlib.compare
+            end) in
+            let queued_delete_conflicts (record : outbox_record) =
+              let earlier = function Some sequence -> sequence < record.sequence | None -> false in
+              if not (Rrbvec.exists (fun (_, _, origin) -> earlier origin) conflict_contexts)
+              then delete_conflict_kinds_between authoritative_before authoritative_after record
+              else
+                Rrbvec.fold_left
+                  (fun conflicts (before, after, origin) ->
+                    if earlier origin then conflicts
+                    else List.fold_left
+                      (fun conflicts kind -> Conflict_set.add kind conflicts)
+                      conflicts (delete_conflict_kinds_between before after record))
+                  Conflict_set.empty conflict_contexts
+                |> Conflict_set.elements
+            in
             let remote_won_queued =
               List.filter_map
                 (fun (record : outbox_record) ->
                    match record.transport_state, record.mutation with
                    | Queued, Types.Delete_blocks { root; _ } ->
                      let after = block_of_database authoritative_after root in
                      let conflicts =
-                       delete_conflict_kinds_between
-                         authoritative_before
-                         authoritative_after
-                         record
+                       queued_delete_conflicts record
                      in
                      if conflicts <> [] && Option.is_some after
                      then Some (record, conflicts)

logseq_overlay_db/lib/types.ml

@@ -562,6 +562,7 @@ type submission_descriptor =
   { mutation_id : Graph.Uuid.t
   ; fingerprint : mutation_fingerprint
   ; state : transport_state
+  ; operation : outliner_operation
   ; dependency_eligible : bool
   ; attempt_count : int
   ; plaintext_bytes : int

logseq_overlay_db/spec/types.mli

@@ -437,6 +437,8 @@ type submission_descriptor =
   { mutation_id : Graph.Uuid.t
   ; fingerprint : mutation_fingerprint
   ; state : transport_state
+  ; operation : outliner_operation
+    (** Durable mutation operation; deletion submissions must be singletons. *)
   ; dependency_eligible : bool
   ; attempt_count : int
   ; plaintext_bytes : int

logseq_overlay_db/test/test_overlay_sync.ml

@@ -4695,6 +4695,76 @@ let database_cases =
     ]
 ;;
 
+(* Sync's pure reducer carries opaque transactions and cannot execute this
+   retirement/rebase decision. Exercise its owning public Database boundary. *)
+let queued_delete_survives_own_append_echo ?(existing_block=false) ?(remote_later=false) ~remote_change db =
+ let behavior="queued delete follows own append echo" in
+ let root=if existing_block then T.authoritative_block_uuid else T.mutation_uuid 9001 in
+ let append=Types.Append_journal {mutation_id=T.mutation_uuid 9002;page=T.mutation_uuid 9003;
+ title="2026-11-07";journal_day=20261107;
+ root={uuid=root;title="Synthetic capture";task_status=None;children=Rrbvec.empty}} in
+ let mutation = if existing_block then
+   Types.Save_block {mutation_id=T.mutation_uuid 9002;block=root;title="Own saved title"}
+   else append in
+ let expected = if existing_block then T.delete_precondition db ~block:root ~behavior
+   else T.empty_precondition ~behavior in
+ let intent=Database.persist_local_intent db ~namespace:(String.make 64 'a') ~payload:"append"
+ ~expected mutation |> T.require_ok ~behavior in
+ let proof=Database.commit_local_intent db intent |> T.require_ok ~behavior in
+ let local=Database.intent_commit_outcome proof in
+ let deletion=Types.Delete_blocks{mutation_id=T.mutation_uuid 9004;root} in
+ ignore(T.commit_mutation db ~expected:(T.delete_precondition db ~block:root ~behavior) deletion ~behavior);
+ let submitted=submit_one db local behavior in
+ let wire=server_transaction ~mutation_id:local.mutation_id ~operation:"append-journal" submitted in
+ let barrier={through=server_cursor 1;checksum=checksum "1111111111111111"} in
+ let sync=Database.inspect_sync db |> T.require_ok ~behavior in
+ let prepared,_=Database.begin_outbox_transition db ~expected:(sync_view_token sync)
+ (Accept_group{batch_id=submission_batch_id submitted;barrier}) |> T.require_ok ~behavior in
+ ignore(Database.apply_outbox_transition db prepared ~encrypted:None |> T.require_ok ~behavior);
+ let transactions = [authoritative_transaction ~cursor:(server_cursor 1)
+   ~transaction:(encoded ~maximum_bytes:65536 wire)] in
+ let transactions = if not remote_change then transactions else
+   transactions @ [authoritative_transaction ~cursor:(server_cursor 2)
+     ~transaction:(encoded (Printf.sprintf
+       "[[\"~:db/add\",[\"~:block/uuid\",\"~u%s\"],\"~:block/updated-at\",1704067200999]]"
+       (Graph.Uuid.to_string root)))] in
+ let batch=authoritative_batch ~maximum_count:16 ~maximum_bytes:65536
+ ~transactions
+ ~through:(server_cursor (if remote_change then 2 else 1)) ~checksum:(Some barrier.checksum) |> T.require_ok ~behavior in
+ let sync=Database.inspect_sync db |> T.require_ok ~behavior in
+ let prepared,crypto=Database.begin_authoritative db ~expected:(sync_view_token sync) batch |> T.require_ok ~behavior in
+ let decrypted=Option.map(fun req->decrypt_submitted_request req behavior)crypto in
+ let commit=match Database.apply_authoritative db prepared ~decrypted |> T.require_ok ~behavior with
+ |Authoritative_applied c->c|_->failwith "deferred" in
+ let remote_won = List.exists (fun terminal -> match terminal.receipt with
+   | Remote_won_receipt r -> Graph.Uuid.equal r.mutation_id (T.mutation_uuid 9004)
+   | _ -> false) commit.terminal_receipts in
+ T.require (remote_won = remote_change) "own echo was mistaken for a remote delete conflict";
+ let sync=Database.inspect_sync db |> T.require_ok ~behavior in
+ T.require (List.length (sync_view_submissions sync) = if remote_change then 0 else 1)
+   "queued delete was lost or genuine remote conflict was ignored";
+ let snapshot=Database.current_snapshot db |> T.require_ok ~behavior in
+ let present=match Database.get_outline_blocks snapshot [root] |> T.require_ok ~behavior with
+   | [Present_block _] -> true | _ -> false in
+ Database.release_snapshot snapshot;
+ T.require (present = remote_change) "own echo resurrected locally deleted block";
+ if remote_later then (
+   let remote = Printf.sprintf
+     "[[\"~:db/add\",[\"~:block/uuid\",\"~u%s\"],\"~:block/updated-at\",1704067200999]]"
+     (Graph.Uuid.to_string root) in
+   let later = commit_plain_authoritative db ~cursor:(server_cursor 2) remote behavior in
+   T.require (List.exists (fun terminal -> match terminal.receipt with
+     | Remote_won_receipt r -> Graph.Uuid.equal r.mutation_id (T.mutation_uuid 9004)
+     | _ -> false) later.terminal_receipts)
+     "real remote change in a later batch did not win")
+ else if not remote_change then (
+   let prepared, request, _ = prepare_submit db [T.mutation_uuid 9004] behavior in
+   let protected = protect request behavior in
+   let sent = Database.apply_outbox_transition db prepared ~encrypted:(Some protected)
+     |> T.require_ok ~behavior in
+   T.require (Option.is_some sent.submission_batch) "preserved delete cannot submit")
+;;
+
 let () =
   Alcotest.run
     "logseq_overlay_db sync"
@@ -4713,6 +4783,15 @@ let () =
       ; Alcotest.test_case "storage failure rolls back mixed batch" `Quick (empty_authoritative_failure_is_atomic ~storage:true)
       ; T.database_case "empty payload keeps both token guards" empty_authoritative_keeps_token_guards
       ; T.database_case "malformed neighbours and empty outer batch reject" empty_authoritative_keeps_invalid_input_rejection ]
+    ; "queued delete own echoes", [
+        T.database_case "queued delete survives own append echo"
+          (queued_delete_survives_own_append_echo ~remote_change:false);
+        T.database_case "queued delete still yields to remote edit after own echo"
+          (queued_delete_survives_own_append_echo ~remote_change:true);
+        T.database_case "queued delete yields to later remote batch after own echo"
+          (queued_delete_survives_own_append_echo ~remote_later:true ~remote_change:false);
+        T.database_case "queued delete survives earlier local save echo"
+          (queued_delete_survives_own_append_echo ~existing_block:true ~remote_change:false)]
     ; "validated wire inputs", pure_cases
     ; "protected same-value rewrites", protected_retraction_cases
     ; "durable sync view", database_cases ]

logseq_sync/lib/pure_reducer/core.ml

@@ -860,6 +860,11 @@ type event =
   | Graph_attachment_failed of scoped_error
   | Sync_inspected of sync_inspection
   | Outbox_transition_applied of outbox_transition_result
+  (** An uncommitted preparation lost its token to a newer durable revision.
+      The exact pending request and a changed token fence recovery; this is not
+      evidence of a commit or permission to repeat an unchanged failed request. *)
+  | Outbox_transition_superseded of
+      { request : outbox_transition_request; sync : Logseq_overlay_db.Types.sync_view }
   | Outbox_transition_failed of outbox_transition_failure
   | Authoritative_batch_applied of authoritative_commit_result
   | Authoritative_batch_deferred of authoritative_deferred_result
@@ -1920,13 +1925,15 @@ let plan_submission core =
            let rec take n acc = function
              | [] -> List.rev acc
              | _ when n = 0 -> List.rev acc
-             | x :: xs -> take (n - 1) (x :: acc) xs
+             | (item : Overlay.submission_descriptor) :: rest ->
+               if item.operation = Overlay.Delete_blocks_operation
+               then if acc = [] then [ item.mutation_id ] else List.rev acc
+               else take (n - 1) (item.mutation_id :: acc) rest
            in
            let ids =
              descriptors
              |> List.filter (fun (item : Overlay.submission_descriptor) ->
                item.state = Overlay.Queued && item.dependency_eligible)
-             |> List.map (fun (item : Overlay.submission_descriptor) -> item.mutation_id)
              |> take core.config.limits.submission_batch_size []
            in
            if ids = []
@@ -3424,6 +3431,28 @@ let step core event =
              ; private_key_package = challenge.private_key_package }) in
          { next; effects = [ publish next; Run request ] }
        | _ -> unchanged core)
+    | Outbox_transition_superseded {request;sync} ->
+      (match core.pending_outbox_transition with
+       | Some pending when graph_scope_is_current core request.scope
+           && pending.scope=request.scope && pending.transition=request.transition
+           && Overlay.sync_token_equal pending.expected request.expected
+           && not(Overlay.sync_token_equal request.expected(Overlay.sync_view_token sync))->
+         (* An independently completed inspection may already have replaced the
+            preparation's view. Tokens are opaque: retain that observation
+            rather than guessing that the superseded result is newer. *)
+         let sync=match core.sync_view with
+           | Some observed when not(Overlay.sync_token_equal request.expected(Overlay.sync_view_token observed))->observed
+           | Some _ | None -> sync in
+         let core={core with sync_view=Some sync;pending_outbox_transition=None} in
+         (match request.transition with
+          | Overlay.Submit_group _ | Retry_group _ -> plan_submission core
+          | Accept_group _ | Reject_group _ ->
+            (* Keep the exact server outcome. Replanning submission here could
+               discard an accepted prefix or repeat already executed work. *)
+            let request={request with expected=Overlay.sync_view_token sync} in
+            {next={core with pending_outbox_transition=Some request};
+             effects=[Delegate(Apply_outbox_transition request)]})
+       | Some _ | None -> unchanged core)
     | Outbox_transition_failed { request; message } ->
       (match core.pending_outbox_transition with
        | Some pending
@@ -4369,6 +4398,7 @@ let event_graph_scope core = function
   | Mirror_inspected (Mirror_available {scope;_}) | Graph_attached {scope;_}
   | Sync_inspected {scope;_} | Outbox_transition_applied {scope;_}
   | Authoritative_batch_applied {scope;_} | Authoritative_batch_deferred {scope;_}
+  | Outbox_transition_superseded {request={scope;_};_}
   | Outbox_transition_failed {request={scope;_};_} -> Some scope
   | Snapshot_activated activation -> Some activation.scope
   | Snapshot_activation_failed error -> Some error.activation.scope
@@ -4550,6 +4580,10 @@ let step core event =
   | Visible_graph_changed _ -> unchanged core
   | Local_cache_deletion_requested id when List.mem id (resident_graphs core) -> unchanged core
   | _ when not core.resident_enabled ->
+    let event = match event with
+      | Local_graph_outbox_changed scope when graph_scope_is_current core scope ->
+        Local_outbox_changed
+      | event -> event in
     let transition=step_graph_policy core event in
     let next={transition.next with resident_next_generation=max core.resident_next_generation transition.next.public_state.snapshot.startup.graph_generation} in
     (match next.restored_residents with

logseq_sync/spec/pure_reducer/core.mli

@@ -646,6 +646,11 @@ type event =
   | Graph_attachment_failed of scoped_error
   | Sync_inspected of sync_inspection
   | Outbox_transition_applied of outbox_transition_result
+  (** An uncommitted preparation lost its token to a newer durable revision.
+      The exact pending request and a changed token fence recovery; this is not
+      evidence of a commit or permission to repeat an unchanged failed request. *)
+  | Outbox_transition_superseded of
+      { request : outbox_transition_request; sync : Logseq_overlay_db.Types.sync_view }
   | Outbox_transition_failed of outbox_transition_failure
   | Authoritative_batch_applied of authoritative_commit_result
   | Authoritative_batch_deferred of authoritative_deferred_result