From b26805bb6ce95b5a6c43aaed0e9909fda533879a Mon Sep 17 00:00:00 2001 From: rcmerci Date: Sat, 7 Nov 2026 16:51:06 +0800 Subject: [PATCH] Fix Capture sync wakeups, deletion replay and preparation races --- .../lib/effect_runner/effect_runner.ml | 22 +- logseq_db_worker/test/test_effect_runner.ml | 229 +++++++++++++++++- logseq_overlay_db/lib/database.ml | 46 +++- logseq_overlay_db/lib/types.ml | 1 + logseq_overlay_db/spec/types.mli | 2 + logseq_overlay_db/test/test_overlay_sync.ml | 79 ++++++ logseq_sync/lib/pure_reducer/core.ml | 38 ++- logseq_sync/spec/pure_reducer/core.mli | 5 + logseq_sync/test/core_contract.ml | 195 +++++++++++++++ .../test/pure_reducer_bad_case_support.ml | 1 + 10 files changed, 608 insertions(+), 10 deletions(-) diff --git a/logseq_db_worker/lib/effect_runner/effect_runner.ml b/logseq_db_worker/lib/effect_runner/effect_runner.ml index 92defbe0..bf44496a 100644 --- a/logseq_db_worker/lib/effect_runner/effect_runner.ml +++ b/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) diff --git a/logseq_db_worker/test/test_effect_runner.ml b/logseq_db_worker/test/test_effect_runner.ml index 798faf38..242c7ded 100644 --- a/logseq_db_worker/test/test_effect_runner.ml +++ b/logseq_db_worker/test/test_effect_runner.ml @@ -3181,10 +3181,237 @@ let test_authoritative_rejection_diagnostics kind () = @ if String.length payload > 30 then [ payload ] else [])))))) ;; +(* The pure reducer can consume an outbox notice, but cannot reproduce a missing + notice from the physical commit owner. Exercise that public runner boundary + against isolated synthetic mirrors, then drain its real events. *) +let test_durable_commit_notifies_own_graph ~resident () = + let module Mutation=Logseq_db_worker_contract.Worker_pure_mutation in + let require condition message=Alcotest.(check bool) message true condition in + T.with_managed(fun fixture-> + let graph_id=Test_support.graph_uuid in + let other_id=Test_support.mutation_uuid 29800 in + List.iter(fun graph_id->ignore(Test_support.seed_mirror ~graph_id fixture.config.application_support_directory)) + (if resident then [graph_id;other_id] else [graph_id]); + Eio_main.run(fun _->Eio.Switch.run(fun sw-> + let graph graph_id : Sync.graph={graph_id;name="Durable notification fixture"; + encrypted=false;schema={major=65;minor=33;exact=true}} in + let state=ref(Core.initial(Core.config ~worker:fixture.config ~sync:(asset_sync_config())) |> Result.get_ok) in + let events=Queue.create() and connections=ref [] and inspections=ref [] in + let authority=ref true and retire_on_projection=ref false in + let post event= + (match event with Core.Projection_push_for_scope _ when !retire_on_projection->authority:=false|_->()); + Queue.add event events in + let sync_runner=Runner.sync_runner ~retire_account:(fun ~account:_->Ok()) ~shutdown:(fun()->()) + ~submit:(function + |Sync.Request(ticket,Sync.Fetch_catalog _)->post(Core.Sync_event(Sync.Runner_completed + (Sync.Completion(ticket,Ok(List.map graph (if resident then [graph_id;other_id] else [graph_id])))))) + |Sync.Request(ticket,Sync.Save_catalog _)->post(Core.Sync_event(Sync.Runner_completed(Sync.Completion(ticket,Ok())))) + |Sync.Start_websocket request->connections:=request.scope :: !connections + |_->())() in + let dependencies=Runner.dependencies ~config:fixture.config ~overlay:fixture.overlay ~sync_runner + ~runtime:(Runner.runtime ~fork:(fun ~sw:_ task->task()) ~sleep:(fun _->()) |> Result.get_ok) + ~publish:(fun _->()) |> Result.get_ok in + let runner=Runner.create ~sw dependencies ~post ~recovery_current:(fun _->false) + ~upload_current:(fun _->false) ~asset_current:(fun _->false) |> Result.get_ok in + let drain ()=while not(Queue.is_empty events)do + let transition=Core.step !state(Queue.take events)in state:=transition.next; + List.iter(fun instruction-> + (match instruction with + |Core.Run_worker(Core.Request(_,Core.Handle_sync_worker_effect(Sync.Inspect_sync scope)))-> + inspections:=scope :: !inspections + |_->()); + Runner.submit runner instruction)transition.effects done in + let drive event=post(Core.Sync_event event);drain() in + let notices ()=Queue.fold(fun scopes->function + |Core.Sync_event(Sync.Local_graph_outbox_changed scope)->scope::scopes + |Core.Sync_event Sync.Local_outbox_changed->Alcotest.fail "durable mutation emitted an unscoped notice" + |_->scopes)[] events in + let intent ?(title="Durable notification") n= + Mutation.create_append ~origin:(Uri.of_string "https://api.logseq.io") ~user:"fixture-user" ~graph:graph_id + ~mutation_id:(Test_support.mutation_uuid n) ~page:(Test_support.mutation_uuid 29801) + ~day:(Logseq_db_worker_contract.Worker_pure_day.create 20261107 |> Result.get_ok) + ~root:{block_id=Test_support.mutation_uuid(n+100);title;task_status=None;children=Rrbvec.empty} + |> Result.get_ok |> fun value->Mutation.Append_intent value in + Fun.protect ~finally:(fun()->Runner.shutdown runner)(fun()-> + drive(Sync.Sign_in_requested{account_generation=0;presentation_generation=0}); + drive(Sync.Account_authenticated{user_id=Some "fixture-user"}); + if resident then ( + drive(Sync.Resident_graphs_changed[graph_id;other_id]); + drive(Sync.Visible_graph_changed other_id)) + else drive(Sync.Graph_selected graph_id); + List.iter(fun connection->drive(Sync.Websocket_opened connection))!connections; + let connection=List.find(fun (value:Sync.connection_scope)->value.graph.graph_id=graph_id)!connections in + let scope=connection.graph in + if resident then require((Core.view !state).sync.snapshot.selected_graph=Some other_id) + "mutation fixture must keep another resident visible"; + inspections:=[]; + let mutate intent=Runner.durable_mutation runner ~scope ~intent ~current:(fun()-> !authority) in + let first=intent 29810 in + let proof=match mutate first with Mutation.Committed proof->proof|_->Alcotest.fail "synthetic append did not commit" in + require(notices()=[scope]) "committed append posts exactly its own scoped sync notice"; + drain(); + require(!inspections=[scope]) "committed append automatically inspects only its own graph"; + require(match mutate first with Mutation.Committed replay->replay=proof|_->false) + "idempotent replay retains original commit proof"; + require(notices()=[scope]) "replayed commit can repair a lost sync notification"; + drain(); + require(match mutate(intent ~title:"Conflicting payload" 29810)with Mutation.Rejected _->true|_->false) + "conflicting intent is rejected"; + require(notices()=[]) "rejected mutation never notifies sync"; + authority:=false; + require(mutate(intent 29811)=Mutation.Unknown) "retired authority cannot commit"; + require(notices()=[]) "unknown mutation never notifies sync"; + authority:=true;retire_on_projection:=true; + require(match mutate(intent 29812)with Mutation.Committed _->true|_->false) + "physical commit retains evidence when presentation retires"; + require(not !authority) "fixture retires authority during actual commit publication"; + require(notices()=[scope]) "caller retirement cannot strand a successfully committed outbox")))) +;; + +let test_durable_commit_during_protection ?(resident=false) mode () = + let module Mutation=Logseq_db_worker_contract.Worker_pure_mutation in + let require condition message=Alcotest.(check bool) message true condition in + T.with_managed(fun fixture-> + let graph_id=Test_support.graph_uuid in + let other_id=Test_support.mutation_uuid 29800 in + List.iter(fun graph_id->ignore(Test_support.seed_mirror ~graph_id fixture.config.application_support_directory)) + (if resident then [graph_id;other_id] else [graph_id]); + Eio_main.run(fun env->Eio.Switch.run(fun sw-> + let graph graph_id : Sync.graph={graph_id;name="Durable notification fixture"; + encrypted=true;schema={major=65;minor=33;exact=true}} in + let state=ref(Core.initial(Core.config ~worker:fixture.config ~sync:(asset_sync_config())) |> Result.get_ok) in + let events=Queue.create() and connections=ref [] and inspections=ref [] in + let authority=ref true and retire_on_projection=ref false in + let held=ref None and running=ref 0 and failures=ref [] and inspected=ref None and closes=ref 0 in + let hold_crypto=ref(mode<>`Ack) and hold_ack=ref(mode=`Ack) and held_ack=ref None in + let sends=ref 0 and superseded=ref 0 and applied=ref None in + let post event= + (match event with Core.Projection_push_for_scope _ when !retire_on_projection->authority:=false|_->()); + Queue.add event events in + let sync_runner=Runner.sync_runner ~retire_account:(fun ~account:_->Ok()) ~shutdown:(fun()->()) + ~submit:(function + |Sync.Request(ticket,Sync.Fetch_catalog _)->post(Core.Sync_event(Sync.Runner_completed + (Sync.Completion(ticket,Ok(List.map graph (if resident then [graph_id;other_id] else [graph_id])))))) + |Sync.Request(ticket,Sync.Save_catalog _)->post(Core.Sync_event(Sync.Runner_completed(Sync.Completion(ticket,Ok())))) + |Sync.Request(ticket,Sync.Load_and_unlock_graph_key scope)-> + post(Core.Sync_event(Sync.Runner_completed(Sync.Completion(ticket,Ok(Sync.graph_key_handle ~id:"synthetic-key" ~scope))))) + |Sync.Request(ticket,Sync.Fetch_and_unlock_graph_key request)-> + post(Core.Sync_event(Sync.Runner_completed(Sync.Completion(ticket,Ok(Sync.graph_key_handle ~id:"synthetic-key" ~scope:request.scope))))) + |Sync.Protected_io(ticket,request)-> + if !hold_crypto then held:=Some(ticket,request) else + let values=match request.Sync.protected_action with Sync.Encrypt_values(_,values)->values|_->assert false in + post(Core.Sync_event(Sync.Runner_completed(Sync.Protected_completion(ticket, + Ok(Sync.Encrypted_values(List.map(fun value->String.make 12 'i',value)values)))))) + |Sync.Send_websocket {message=Logseq_sync_pure_reducer.Sync_protocol.Client.Tx_batch _;_}->incr sends + |Sync.Close_websocket _->incr closes + |Sync.Start_websocket request->connections:=request.scope :: !connections + |_->())() in + let dependencies=Runner.dependencies ~config:fixture.config ~overlay:fixture.overlay ~sync_runner + ~runtime:(Runner.runtime ~fork:(fun ~sw task->incr running;Eio.Fiber.fork ~sw(fun()->Fun.protect ~finally:(fun()->decr running)task)) ~sleep:(fun _->()) |> Result.get_ok) + ~publish:(fun _->()) |> Result.get_ok in + let runner=Runner.create ~sw dependencies ~post ~recovery_current:(fun _->false) + ~upload_current:(fun _->false) ~asset_current:(fun _->false) |> Result.get_ok in + let drain ()=while not(Queue.is_empty events)do + let event=Queue.take events in + (match event with + |Core.Runner_completed(Core.Sync_worker_effect_completed(_,Error error))->failures:=error:: !failures + |Core.Runner_completed(Core.Sync_worker_effect_completed(_,Ok{event=Some(Sync.Sync_inspected{sync;_});_}))->inspected:=Some sync + |Core.Runner_completed(Core.Sync_worker_effect_completed(_,Ok{event=Some(Sync.Outbox_transition_superseded _);_}))->incr superseded + |Core.Runner_completed(Core.Sync_worker_effect_completed(_,Ok{event=Some(Sync.Outbox_transition_applied result);_}))->applied:=Some result + |_->()); + let transition=Core.step !state event in state:=transition.next; + List.iter(fun instruction-> + (match instruction with + |Core.Run_worker(Core.Request(_,Core.Handle_sync_worker_effect(Sync.Inspect_sync scope)))-> + inspections:=scope :: !inspections + |_->()); + match instruction with + |Core.Run_worker(Core.Request(_,Core.Handle_sync_worker_effect(Sync.Apply_outbox_transition request))) + when !hold_ack && (match request.transition with Logseq_overlay_db.Types.Accept_group _->true|_->false)-> + held_ack:=Some(instruction,request) + |_->Runner.submit runner instruction)transition.effects done in + let until condition=Eio.Time.with_timeout_exn(Eio.Stdenv.clock env)3.(fun()-> + while not(condition())do drain();Eio.Fiber.yield()done;drain()) in + let settle ()=until(fun()-> !running=0 && Queue.is_empty events) in + let drive event=post(Core.Sync_event event);settle() in + let intent ?(title="Durable notification") n= + Mutation.create_append ~origin:(Uri.of_string "https://api.logseq.io") ~user:"fixture-user" ~graph:graph_id + ~mutation_id:(Test_support.mutation_uuid n) ~page:(Test_support.mutation_uuid 29801) + ~day:(Logseq_db_worker_contract.Worker_pure_day.create 20261107 |> Result.get_ok) + ~root:{block_id=Test_support.mutation_uuid(n+100);title;task_status=None;children=Rrbvec.empty} + |> Result.get_ok |> fun value->Mutation.Append_intent value in + Fun.protect ~finally:(fun()->Runner.shutdown runner)(fun()-> + drive(Sync.Sign_in_requested{account_generation=0;presentation_generation=0}); + drive(Sync.Account_authenticated{user_id=Some "fixture-user"}); + if resident then ( + drive(Sync.Resident_graphs_changed[graph_id;other_id]); + drive(Sync.Visible_graph_changed other_id)) + else drive(Sync.Graph_selected graph_id); + List.iter(fun connection->drive(Sync.Websocket_opened connection))!connections; + let connection=List.find(fun (value:Sync.connection_scope)->value.graph.graph_id=graph_id)!connections in + let scope=connection.graph in + if resident then require((Core.view !state).sync.snapshot.selected_graph=Some other_id) + "mutation fixture must keep another resident visible"; + inspections:=[]; + let mutate intent=Runner.durable_mutation runner ~scope ~intent ~current:(fun()-> !authority) in + require(match mutate(intent 29810)with Mutation.Committed _->true|_->false) "first append commits"; + (match mode with + |`Ack-> + settle();require(!sends=1) "first capture reaches actual submission"; + drive(Sync.Websocket_message(connection,Logseq_sync_pure_reducer.Sync_protocol.Server.Tx_batch_ok + {t=Logseq_db_types.Server_cursor.of_int64 1L |> Result.get_ok;checksum=Some "0123456789abcdef"})); + let instruction,request=Option.get !held_ack in + require(match mutate(intent 29811)with Mutation.Committed _->true|_->false) "second append commits before ACK preparation starts"; + settle();hold_ack:=false;Runner.submit runner instruction;settle(); + let result=Option.get !applied in + require(result.commit.transition=request.transition) "ACK recovery preserves its exact server barrier"; + let records=Logseq_overlay_db.Types.sync_view_submissions result.sync in + require(List.length records=2 && List.exists(fun (item:Logseq_overlay_db.Types.submission_descriptor)-> + match item.state with Accepted_pending_authoritative _->true|_->false)records) + "ACK settles only its submitted capture while later capture stays durable" + |`Crypto|`Malformed|`Crypto_error-> + until(fun()-> !held<>None); + require(!running>0) "first outbox preparation is suspended in protected IO"; + if mode=`Crypto then ( + require(match mutate(intent 29811)with Mutation.Committed _->true|_->false) "second append commits while protection waits"; + until(fun()-> !running=1 && Queue.is_empty events); + let queued=Logseq_overlay_db.Types.sync_view_submissions(Option.get !inspected) in + require(List.length queued=2 && List.for_all(fun (item:Logseq_overlay_db.Types.submission_descriptor)->item.state=Queued)queued) + "both rapid captures remain durably queued before crypto release"); + let ticket,request=Option.get !held in + let encrypted=match request.Sync.protected_action with + |Sync.Encrypt_values(_,values)->List.map(fun value->String.make 12 'i',value)values + |_->Alcotest.fail "expected encryption request" in + hold_crypto:=false; + let result=match mode with + |`Crypto_error->Error(Sync.Crypto_failed(Sync.Invalid_key_material,"synthetic invalid key")) + |`Malformed->Ok(Sync.Encrypted_values []) + |_->Ok(Sync.Encrypted_values encrypted) in + post(Core.Sync_event(Sync.Runner_completed(Sync.Protected_completion(ticket,result)))); + settle()); + if mode=`Malformed || mode=`Crypto_error then ( + require(List.length !failures=1 && !closes=1) "genuine crypto failure remains visible and closes connection"; + require(!superseded=0 && !sends=0) "genuine crypto failure is never retried or sent") + else ( + require(!failures=[] && !closes=0) "concurrent durable commit recovers without failing or closing sync"; + require(!superseded=1) "exactly one revision-driven recovery"; + require(!sends=1) "capture submits once after recovery"); + require((Core.view !state).graph.phase=Core.Graph_open) "concurrent durable commit preserves attached graph")))) +;; + let () = Alcotest.run "logseq db worker effect runner" - [ ( "authoritative diagnostics", [ + [ ( "durable protection race", [ + Alcotest.test_case "second append during protected preparation" `Quick (test_durable_commit_during_protection `Crypto); + Alcotest.test_case "second append before ACK preparation" `Quick (test_durable_commit_during_protection `Ack); + Alcotest.test_case "genuine crypto failure is not retried" `Quick (test_durable_commit_during_protection `Crypto_error); + Alcotest.test_case "malformed crypto result is not retried" `Quick (test_durable_commit_during_protection `Malformed); + Alcotest.test_case "hidden resident protection race" `Quick (test_durable_commit_during_protection ~resident:true `Crypto)]) + ; ( "durable sync notification", [ + Alcotest.test_case "single graph commit and replay" `Quick (test_durable_commit_notifies_own_graph ~resident:false); + Alcotest.test_case "hidden resident commit and replay" `Quick (test_durable_commit_notifies_own_graph ~resident:true) ]) + ; ( "authoritative diagnostics", [ Alcotest.test_case "real own echo mismatch retains safe integrity reason" `Quick (test_authoritative_rejection_diagnostics `Integrity); Alcotest.test_case "real cursor rejection reaches Diagnostics" `Quick (test_authoritative_rejection_diagnostics `Cursor); Alcotest.test_case "real format rejection reaches Diagnostics" `Quick (test_authoritative_rejection_diagnostics `Format); diff --git a/logseq_overlay_db/lib/database.ml b/logseq_overlay_db/lib/database.ml index 5a286969..c36ac2f3 100644 --- a/logseq_overlay_db/lib/database.ml +++ b/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,6 +10106,46 @@ 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) -> @@ -10112,10 +10153,7 @@ let commit_authoritative_candidate database prepared = | 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) diff --git a/logseq_overlay_db/lib/types.ml b/logseq_overlay_db/lib/types.ml index 30344134..d10f55b7 100644 --- a/logseq_overlay_db/lib/types.ml +++ b/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 diff --git a/logseq_overlay_db/spec/types.mli b/logseq_overlay_db/spec/types.mli index 1b5acf6c..72946aec 100644 --- a/logseq_overlay_db/spec/types.mli +++ b/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 diff --git a/logseq_overlay_db/test/test_overlay_sync.ml b/logseq_overlay_db/test/test_overlay_sync.ml index b7aea549..0d5d5365 100644 --- a/logseq_overlay_db/test/test_overlay_sync.ml +++ b/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 ] diff --git a/logseq_sync/lib/pure_reducer/core.ml b/logseq_sync/lib/pure_reducer/core.ml index 6e35db66..ebfd6f08 100644 --- a/logseq_sync/lib/pure_reducer/core.ml +++ b/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 diff --git a/logseq_sync/spec/pure_reducer/core.mli b/logseq_sync/spec/pure_reducer/core.mli index 696d6441..a2e5e4bf 100644 --- a/logseq_sync/spec/pure_reducer/core.mli +++ b/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 diff --git a/logseq_sync/test/core_contract.ml b/logseq_sync/test/core_contract.ml index dceb8070..885643b3 100644 --- a/logseq_sync/test/core_contract.ml +++ b/logseq_sync/test/core_contract.ml @@ -537,6 +537,7 @@ let warm_encrypted_graph_loads_key_before_attach_and_protects_outbox () = { mutation_id ; fingerprint ; state = Logseq_overlay_db.Types.Queued + ; operation = Logseq_overlay_db.Types.Save_block_operation ; dependency_eligible = true ; attempt_count = 0 ; plaintext_bytes = 16 @@ -1356,6 +1357,7 @@ let submitted_core () = ^ Logseq_db_types.Graph_types.Uuid.to_string mutation_id) |> Result.get_ok ; state = Overlay.Queued + ; operation = Overlay.Save_block_operation ; dependency_eligible = true ; attempt_count = 0 ; plaintext_bytes = 16 @@ -1904,6 +1906,7 @@ let canonical_overlay_happy_path () = { mutation_id ; fingerprint ; state = Overlay.Queued + ; operation = Overlay.Save_block_operation ; dependency_eligible = true ; attempt_count = 0 ; plaintext_bytes = 16 @@ -2732,6 +2735,7 @@ let submitted_delete_recovery_fixture () = Overlay.Mutation_fingerprint.of_string "mutation-fingerprint:v1:delete-recovery" |> Result.get_ok ; state = Overlay.Queued + ; operation = Overlay.Delete_blocks_operation ; dependency_eligible = true ; attempt_count = 0 ; plaintext_bytes = 128 @@ -4056,6 +4060,7 @@ let rejected_submission_feedback () = ^ Logseq_db_types.Graph_types.Uuid.to_string mutation_id) |> Result.get_ok ; state = Overlay.Blocked + ; operation = Overlay.Save_block_operation ; dependency_eligible = false ; attempt_count = 1 ; plaintext_bytes = 16 @@ -6721,3 +6726,193 @@ let scenarios = scenarios @ [ Alcotest.test_case "cache decoder keeps safe field cause" `Quick cache_decoder_keeps_safe_field_cause; Alcotest.test_case "snapshot metadata keeps safe field cause" `Quick snapshot_metadata_keeps_safe_field_cause] ;; + +(* The sync reducer owns batching: descriptors arrive through its public inspected + view, rather than injecting a rejected storage transition. *) +let submission_operation_batches operations expected_lengths () = + let descriptors = + List.mapi (fun index operation -> + { Overlay.mutation_id = + Logseq_db_types.Graph_types.Uuid.of_string + (Printf.sprintf "00000000-0000-4000-8000-%012d" (index + 1)) + |> Result.get_ok + ; fingerprint = token Overlay.Mutation_fingerprint.of_string + (Printf.sprintf "mutation-fingerprint:v1:batch-%d" index) + ; operation + ; state = Overlay.Queued + ; dependency_eligible = true + ; attempt_count = 0 + ; plaintext_bytes = 16 + ; protected_bytes = Some 16 + }) operations + in + let sync ordinal submissions = recovery_sync () ordinal submissions in + let core, connection = restore_recovery_core (sync 0 []) in + let inspecting = Core.step core Core.Local_outbox_changed in + let first_event = Core.Sync_inspected + { scope = connection.graph; sync = sync 1 descriptors } in + let first = Core.step inspecting.next first_event in + ignore (check_replay "operation batch" inspecting.next first_event first); + let rec drain (planned : Core.transition) ordinal checkpoint remaining = function + | [] -> + Alcotest.(check int) "no stranded mutations" 0 (List.length remaining); + require_no_new_submission planned.effects + | count :: counts -> + let request = recovery_request planned.effects in + let rec split n acc xs = match n, xs with + | 0, _ | _, [] -> List.rev acc, xs + | _, x :: tail -> split (n - 1) (x :: acc) tail in + let expected, rest = split count [] remaining in + Alcotest.(check bool) "ordered contiguous submission" + true (request.transition = Overlay.Submit_group + (List.map (fun (item : Overlay.submission_descriptor) -> item.mutation_id) expected)); + let batch_id = token Overlay.Submission_batch_id.of_string + (Printf.sprintf "submission-batch:v1:operation-%d" ordinal) in + let wires = List.map (fun (item : Overlay.submission_descriptor) -> + Overlay.submission_wire ~maximum_bytes:1024 ~mutation_id:item.mutation_id + ~operation:item.operation ~protected_transaction:"[]" |> Result.get_ok) expected in + let batch = Overlay.submission_batch ~maximum_wires:32 ~maximum_bytes:4096 + ~batch_id ~t_before:(recovery_cursor checkpoint) ~wires |> Result.get_ok in + let view state ordinal = recovery_sync () ~checkpoint ordinal + (List.map (fun (item : Overlay.submission_descriptor) -> {item with state}) expected @ rest) in + let submitted = Core.step planned.next + (recovery_commit request (view (Overlay.Submitted batch_id) (ordinal + 1)) (Some batch)) in + let through = checkpoint + count in + let acknowledged = Core.step submitted.next + (Core.Websocket_message (connection, Protocol.Server.Tx_batch_ok + {t = server_cursor_of_int through; checksum = Some "0123456789abcdef"})) in + let accept = recovery_request acknowledged.effects in + let accepted = Core.step acknowledged.next + (recovery_commit accept + (view (Overlay.Accepted_pending_authoritative batch_id) (ordinal + 2)) None) in + require_no_new_submission accepted.effects; + let pulled = Core.step accepted.next + (Core.Websocket_message (connection, Protocol.Server.Pull_ok + { t = server_cursor_of_int through; checksum = Some "0123456789abcdef" + ; txs = List.init count (fun i -> Protocol.Server. + {t = server_cursor_of_int (checkpoint + i + 1); tx = "[]"; outliner_op = None}) })) in + let applied = recovery_apply pulled.next connection + (recovery_sync () ~checkpoint:through (ordinal + 3) rest) in + drain applied (ordinal + 4) through rest counts + in + drain first 1 0 descriptors expected_lengths +;; + +let scenarios = scenarios @ + let open Overlay in + let normal = Append_journal_operation and delete = Delete_blocks_operation in + List.map (fun (name, operations, groups) -> + Alcotest.test_case name `Quick (submission_operation_batches operations groups)) + [ "delete first stays singleton", [delete; normal; normal], [1; 2] + ; "delete middle preserves normal batches", [normal; normal; delete; normal; normal], [2; 1; 2] + ; "delete last stays singleton", [normal; normal; delete], [2; 1] + ; "two deletes drain as singletons", [delete; delete], [1; 1] + ; "normal mutations retain batching", [Save_block_operation; Insert_blocks_operation; + Append_journal_operation; Create_journal_page_operation; Set_task_status_operation; + Clear_task_status_operation; Publish_asset_operation], [7] + ; "normal batch count cap remains", List.init 34 (fun _ -> normal), [32; 2] + ; "delete at batch cap remains singleton", List.init 32 (fun _ -> normal) @ [delete; normal], [32; 1; 1] + ] +;; + +let nonresident_scoped_outbox_notice () = + let core, connection = restore_recovery_core (recovery_sync () 0 []) in + let scope = connection.graph in + let changed = Core.step core (Core.Local_graph_outbox_changed scope) in + check_instructions "matching scoped notice" [Core.Delegate (Core.Inspect_sync scope)] changed.effects; + List.iter (fun stale -> + let event = Core.Local_graph_outbox_changed stale in + let ignored = Core.step core event in + check_instructions "stale scoped notice" [] ignored.effects; + check_observed "stale scoped notice" "state unchanged" (observe core) (observe ignored.next)) + [ {scope with graph_id = other_graph_id} + ; {scope with graph_generation = scope.graph_generation + 1} + ; {scope with account = {scope.account with account_generation = scope.account.account_generation + 1}} + ; {scope with account = {scope.account with lifecycle_generation = Int64.succ scope.account.lifecycle_generation}} ] +;; +let scenarios = scenarios @ [Alcotest.test_case "nonresident scoped outbox notice is fenced" + `Quick nonresident_scoped_outbox_notice] +;; + +let submission_keeps_dependency_eligibility () = + let make index operation dependency_eligible : Overlay.submission_descriptor = + { mutation_id = Logseq_db_types.Graph_types.Uuid.of_string + (Printf.sprintf "00000000-0000-4000-8000-%012d" index) |> Result.get_ok + ; fingerprint = token Overlay.Mutation_fingerprint.of_string + (Printf.sprintf "mutation-fingerprint:v1:dependency-%d" index) + ; operation; dependency_eligible; state = Overlay.Queued; attempt_count = 0 + ; plaintext_bytes = 16; protected_bytes = Some 16 } + in + let ineligible = make 1 Overlay.Delete_blocks_operation false in + let eligible = make 2 Overlay.Append_journal_operation true in + List.iter (fun (descriptors, expected) -> + let core, connection = restore_recovery_core (recovery_sync () 0 []) in + let inspecting = Core.step core Core.Local_outbox_changed in + let planned = Core.step inspecting.next (Core.Sync_inspected + {scope = connection.graph; sync = recovery_sync () 1 descriptors}) in + match expected with + | [] -> require_no_new_submission planned.effects + | ids -> Alcotest.(check bool) "ineligible descriptors are never submitted" + true ((recovery_request planned.effects).transition = Overlay.Submit_group ids)) + [ [ineligible], []; [ineligible; eligible], [eligible.mutation_id] ] +;; +let scenarios = scenarios @ [Alcotest.test_case "submission respects dependency eligibility" + `Quick submission_keeps_dependency_eligibility] +;; + +let superseded_outbox_request_recovers kind () = + let first=List.hd mutation_ids in + let descriptor state : Overlay.submission_descriptor = + {mutation_id=first;fingerprint=token Overlay.Mutation_fingerprint.of_string "mutation-fingerprint:v1:superseded"; + operation=Overlay.Append_journal_operation;state;dependency_eligible=true;attempt_count=0; + plaintext_bytes=16;protected_bytes=Some 16} in + let pending = match kind with + | `Submit | `Retry -> + let state=match kind with `Retry->Overlay.Submitted(token Overlay.Submission_batch_id.of_string "submission-batch:v1:superseded")|_->Overlay.Queued in + let core,connection=restore_recovery_core(recovery_sync() 80 []) in + let inspecting=Core.step core Core.Local_outbox_changed in + Core.step inspecting.next(Core.Sync_inspected{scope=connection.graph;sync=recovery_sync() 81 [descriptor state]}) + | `Accept | `Reject -> + let core,connection=submitted_core()in + let message=match kind with + |`Accept->Protocol.Server.Tx_batch_ok{t=server_cursor_of_int 3;checksum=Some "0123456789abcdef"} + |_->Protocol.Server.Tx_reject{reason=Db_transact_failed;t=Some(server_cursor_of_int 1); + checksum=Some "0123456789abcdef";success_tx_ids=[first];failed_tx_id=Some(List.nth mutation_ids 1); + missing_block_uuids=[];error_detail=None;data=None} in + Core.step core(Core.Websocket_message(connection,message)) in + let request=recovery_request pending.effects in + let fresh=recovery_sync() 82 [descriptor Overlay.Queued] in + let superseded request sync=Core.Outbox_transition_superseded{request;sync} in + let changed=Core.step pending.next(superseded request fresh) in + let retry=recovery_request changed.effects in + Alcotest.(check bool) "recovery advances only to observed fresh token" true + (Overlay.sync_token_equal retry.expected(Overlay.sync_view_token fresh)); + Alcotest.(check bool) "same graph and key retained" true (retry.scope=request.scope && retry.key=request.key); + (match kind with + |`Accept|`Reject->Alcotest.(check bool) "exact validated server outcome retained" true (retry.transition=request.transition) + |`Submit|`Retry->Alcotest.(check bool) "submission replans from current queued descriptors" true + (retry.transition=Overlay.Submit_group[first])); + check_instructions "superseded completion cannot replay after token advances" [] + (Core.step changed.next(superseded request fresh)).effects; + let same_token=Overlay.sync_view ~token:request.expected ~checkpoint:(recovery_cursor 0) ~submissions:[] in + check_instructions "unchanged token never retries" [] (Core.step pending.next(superseded request same_token)).effects; + List.iter(fun stale->check_instructions "wrong pending owner ignored" [] + (Core.step pending.next(superseded stale fresh)).effects) + [{request with scope={request.scope with graph_generation=request.scope.graph_generation+1}}; + {request with scope={request.scope with account={request.scope.account with lifecycle_generation=Int64.succ request.scope.account.lifecycle_generation}}}; + {request with expected=Overlay.sync_view_token fresh}; + {request with transition=Overlay.Submit_group[]}]; + let inspecting=Core.step pending.next Core.Local_outbox_changed in + let latest=recovery_sync() 83 [descriptor Overlay.Queued] in + let observed=Core.step inspecting.next(Core.Sync_inspected{scope=request.scope;sync=latest}) in + let recovered=Core.step observed.next(superseded request fresh) in + Alcotest.(check bool) "late superseded result cannot replace independently refreshed sync view" true + (Overlay.sync_token_equal(recovery_request recovered.effects).expected(Overlay.sync_view_token latest)); + let signed_out=Core.step pending.next(Core.Account_authenticated{user_id=None})in + check_instructions "retired account ignores superseded preparation" [] + (Core.step signed_out.next(superseded request fresh)).effects +;; +let scenarios=scenarios @ List.map(fun(name,kind)->Alcotest.test_case name `Quick(superseded_outbox_request_recovers kind)) + ["superseded submit replans and fences",`Submit;"superseded retry replans and fences",`Retry; + "superseded accept preserves outcome",`Accept;"superseded reject preserves partition",`Reject] +;; diff --git a/logseq_sync/test/pure_reducer_bad_case_support.ml b/logseq_sync/test/pure_reducer_bad_case_support.ml index 0993009a..00dc0e92 100644 --- a/logseq_sync/test/pure_reducer_bad_case_support.ml +++ b/logseq_sync/test/pure_reducer_bad_case_support.ml @@ -184,6 +184,7 @@ let queued_sync sync_token checkpoint mutation_id = { mutation_id ; fingerprint ; state = Overlay.Queued + ; operation = Overlay.Save_block_operation ; dependency_eligible = true ; attempt_count = 0 ; plaintext_bytes = 16