diff --git a/docs/agent-guide/implemented/architecture/2026-10-09-server-cursor-int64.md b/docs/agent-guide/implemented/architecture/2026-10-09-server-cursor-int64.md new file mode 100644 index 0000000..0dd2961 --- /dev/null +++ b/docs/agent-guide/implemented/architecture/2026-10-09-server-cursor-int64.md @@ -0,0 +1,107 @@ +# Server Cursor Int64 + +## Problem + +Authoritative server transaction progress is represented by unrelated integers and +versioned strings. The string comparison rejects normally ordered pull batches +across decimal digit boundaries, and repeated conversions do not enforce numeric +range or non-negativity consistently. The server JSON number representation also +requires an explicit JavaScript safe-integer bound. + +## Decision + +Own one abstract, non-negative int64 cursor in logseq_db_types.Server_cursor, +the existing common dependency of storage and overlay. Expose the identical +abstract type through logseq_overlay_db.Server_cursor and Types.Server_cursor. +The overlay spec signature keeps the zero/of_int64/to_int64/equal/compare API, +with type identity tied to the common abstract type, never to public int64. + +Migrate authoritative progress in protocol, reducer, worker, snapshot/checkpoint, +overlay and persistence to this type. Convert to/from int64 only in I/O codecs +and checked interval arithmetic. JSON numbers must be in [0, 9007199254740991]. +Preserve existing numeric versioned cursor tokens only inside legacy persistence +codecs. Do not migrate local changefeed cursors, Datascript transaction entity IDs, +submission counts/ordinals, graph or lifecycle generations. + +The only currently identified necessary Dune change is in +logseq_overlay_db/spec/dune: + + (modules server_cursor types database) + (virtual_modules server_cursor types database) + +Both shared types and the overlay implementation discover the new implementation +modules through their existing standard module selection. No new package or +storage-to-overlay dependency is needed. Further necessary Dune changes require +separate explicit permission. + + +Adopt the existing common db_types dependency as the owner of the abstract +non-negative int64 cursor and expose the identical type through overlay. +The user authorized the migration and exactly the two spec Dune module-list +changes above. Other Dune files remain unchanged. + +## Alternatives considered + +### Define the type only in overlay + +Storage already sits below the overlay implementation, so using an overlay-owned +cursor in the storage checkpoint API introduces a package implementation cycle. + +### Keep storage and protocol as raw int64 + +This permits unconstrained values outside I/O and fails the requested shared +progress identity. Only codec boundaries should expose raw numeric values. + +### Retain VERSIONED_TOKEN and change its comparator + +This keeps unrelated token factories and malformed numeric states constructible, +and does not unify protocol/checkpoint progress or protect numeric ranges. + +## Acceptance criteria + +- Normal public reducer pull batches crossing 9 to 10 and 99 to 100 are admitted; + duplicates and reverse order are rejected. +- Cursor construction rejects negatives; I/O rejects overflow and values beyond + the server's JSON safe-integer range without silent narrowing. +- SQLite checkpoints and existing outbox/receipt tokens recover the same progress. +- Direct production consumers share the same abstract type, with no public raw + integer/string constructors bypassing of_int64 validation. +- Relevant tests, all direct consumer builds, independent diff review and exact + PR head CI complete; no merge or deployment is performed. + +## Risks + +- Legacy persistence syntax remains a compatibility boundary rather than a public + domain constructor; malformed legacy numeric values will be rejected explicitly. +- Non-negative int64 is a broader domain than the JavaScript wire can represent; + wire codecs must impose the additional upper bound. +- Existing arithmetic that adds ordinals must reject int64 overflow rather than + wrap; full domain values must not pass through OCaml int conversions. +- This change does not address the independent WebSocket reconnect finding. + +## Consequences + +Authoritative progress now has one abstract identity across storage, overlay, +protocol, reducer and worker; negative values cannot enter domain state. +Legacy numeric token bytes remain unchanged at the persistence boundary, and +wire encoders/decoders impose the server's JavaScript safe-integer ceiling. +Checked interval arithmetic rejects an overflowing submission before freezing or +persisting it, including an independent suffix after a definitive rejection. +The native UI consumers build without changing their cursor display semantics. + +The HTTP snapshot baseline retains its existing 64 MiB response budget, shared +between its transport request and decoder. WebSocket response decoding retains +its independent 262,144-byte budget. A public Core.step bootstrap regression uses +ordinary cursor values and a valid larger HTTP pull response; decimal-boundary +bug regressions remain exclusively at that public reducer boundary. + +The migration does not run live account or Simulator acceptance. Repository-wide +decision validation still reports three pre-existing unrelated document errors; +this decision is independently validated. + +## Questions + +- May logseq_overlay_db/spec/dune add server_cursor to its modules and + virtual_modules lists as shown above? AGENTS.md prohibits Dune edits unless + explicitly requested. The user explicitly approved exactly these two lines on + 2026-10-09; no other Dune modification is authorized. diff --git a/logseq_db_storage/lib/sync_checkpoint_store.ml b/logseq_db_storage/lib/sync_checkpoint_store.ml index e5f6b89..60cfbd5 100644 --- a/logseq_db_storage/lib/sync_checkpoint_store.ml +++ b/logseq_db_storage/lib/sync_checkpoint_store.ml @@ -37,7 +37,11 @@ let initialize_database db checkpoint = (Graph_types.Uuid.to_string checkpoint.graph_id)); Sqlite3.Rc.check (Sqlite3.bind_int statement 3 checkpoint.schema.major); Sqlite3.Rc.check (Sqlite3.bind_int statement 4 checkpoint.schema.minor); - Sqlite3.Rc.check (Sqlite3.bind_int statement 5 checkpoint.applied_server_t); + Sqlite3.Rc.check + (Sqlite3.bind_int64 + statement + 5 + (Server_cursor.to_int64 checkpoint.applied_server_t)); Sqlite3.Rc.check (Sqlite3.bind_text statement 6 checkpoint.checksum); Sqlite3.Rc.check (Sqlite3.bind_text @@ -104,7 +108,14 @@ let read_database db = let graph_id = Sqlite3.column_text statement 1 in let major = Sqlite3.column_int statement 2 in let minor = Sqlite3.column_int statement 3 in - let applied_server_t = Sqlite3.column_int statement 4 in + let applied_server_t = + match Sqlite3.column statement 4 with + | Sqlite3.Data.INT value -> + Server_cursor.of_int64 value + |> Result.map_error (fun `Negative_cursor -> + "sync_meta cursor is negative") + | _ -> Error "sync_meta cursor is not an integer" + in let checksum = Sqlite3.column_text statement 5 in let status = match Sqlite3.column statement 6 with @@ -124,9 +135,10 @@ let read_database db = match Graph_types.Uuid.of_string graph_id with | Error _ -> Error "sync_meta graph id is invalid" | Ok graph_id -> - (match status, last_error with - | Error message, _ | _, Error message -> Error message - | Ok status, Ok last_error -> + (match applied_server_t, status, last_error with + | Error message, _, _ | _, Error message, _ | _, _, Error message -> + Error message + | Ok applied_server_t, Ok status, Ok last_error -> (match Sync_checkpoint.create_full ~graph_id @@ -169,7 +181,11 @@ let update_database db checkpoint = (Graph_types.Uuid.to_string checkpoint.graph_id)); Sqlite3.Rc.check (Sqlite3.bind_int statement 3 checkpoint.schema.major); Sqlite3.Rc.check (Sqlite3.bind_int statement 4 checkpoint.schema.minor); - Sqlite3.Rc.check (Sqlite3.bind_int statement 5 checkpoint.applied_server_t); + Sqlite3.Rc.check + (Sqlite3.bind_int64 + statement + 5 + (Server_cursor.to_int64 checkpoint.applied_server_t)); Sqlite3.Rc.check (Sqlite3.bind_text statement 6 checkpoint.checksum); Sqlite3.Rc.check (Sqlite3.bind_text diff --git a/logseq_db_types/lib/limits.ml b/logseq_db_types/lib/limits.ml index 66858ca..16816b4 100644 --- a/logseq_db_types/lib/limits.ml +++ b/logseq_db_types/lib/limits.ml @@ -1,4 +1,5 @@ let maximum_request_bytes = 1_048_576 let maximum_response_bytes = 262_144 +let maximum_snapshot_baseline_response_bytes = 64 * 1024 * 1024 let maximum_push_bytes = 65_536 let maximum_changed_uuids = 1_024 diff --git a/logseq_db_types/lib/limits.mli b/logseq_db_types/lib/limits.mli index 4704819..aec52bc 100644 --- a/logseq_db_types/lib/limits.mli +++ b/logseq_db_types/lib/limits.mli @@ -1,4 +1,5 @@ val maximum_request_bytes : int val maximum_response_bytes : int +val maximum_snapshot_baseline_response_bytes : int val maximum_push_bytes : int val maximum_changed_uuids : int diff --git a/logseq_db_types/lib/server_cursor.ml b/logseq_db_types/lib/server_cursor.ml new file mode 100644 index 0000000..b8c977d --- /dev/null +++ b/logseq_db_types/lib/server_cursor.ml @@ -0,0 +1,7 @@ +type t = int64 + +let zero = 0L +let of_int64 value = if value < 0L then Error `Negative_cursor else Ok value +let to_int64 value = value +let equal = Int64.equal +let compare = Int64.compare diff --git a/logseq_db_types/lib/server_cursor.mli b/logseq_db_types/lib/server_cursor.mli new file mode 100644 index 0000000..b7e7506 --- /dev/null +++ b/logseq_db_types/lib/server_cursor.mli @@ -0,0 +1,8 @@ +(** A non-negative authoritative server transaction cursor. *) +type t + +val zero : t +val of_int64 : int64 -> (t, [ `Negative_cursor ]) result +val to_int64 : t -> int64 +val equal : t -> t -> bool +val compare : t -> t -> int diff --git a/logseq_db_types/lib/sync_checkpoint.ml b/logseq_db_types/lib/sync_checkpoint.ml index 3f78827..45fc910 100644 --- a/logseq_db_types/lib/sync_checkpoint.ml +++ b/logseq_db_types/lib/sync_checkpoint.ml @@ -6,7 +6,7 @@ type t = { format_version : int ; graph_id : Graph_types.Uuid.t ; schema : Graph_types.schema_version - ; applied_server_t : int + ; applied_server_t : Server_cursor.t ; checksum : string ; status : status ; last_error : string option @@ -26,8 +26,6 @@ let valid_checksum value = let create_full ~graph_id ~schema ~applied_server_t ~checksum ~status ~last_error = if schema.Graph_types.major < 0 || schema.minor < 0 then Error "sync schema version must be non-negative" - else if applied_server_t < 0 - then Error "applied server t must be non-negative" else if not (valid_checksum checksum) then Error "sync checksum must be 16 lowercase hexadecimal characters" else if status = Active && Option.is_some last_error @@ -61,7 +59,7 @@ let equal left right = left.format_version = right.format_version && Graph_types.Uuid.equal left.graph_id right.graph_id && left.schema = right.schema - && left.applied_server_t = right.applied_server_t + && Server_cursor.equal left.applied_server_t right.applied_server_t && String.equal left.checksum right.checksum && left.status = right.status && left.last_error = right.last_error diff --git a/logseq_db_types/lib/sync_checkpoint.mli b/logseq_db_types/lib/sync_checkpoint.mli index f63bff1..09c8cc5 100644 --- a/logseq_db_types/lib/sync_checkpoint.mli +++ b/logseq_db_types/lib/sync_checkpoint.mli @@ -6,7 +6,7 @@ type t = { format_version : int ; graph_id : Graph_types.Uuid.t ; schema : Graph_types.schema_version - ; applied_server_t : int + ; applied_server_t : Server_cursor.t ; checksum : string ; status : status ; last_error : string option @@ -17,14 +17,14 @@ val format_version : int val create : graph_id:Graph_types.Uuid.t -> schema:Graph_types.schema_version - -> applied_server_t:int + -> applied_server_t:Server_cursor.t -> checksum:string -> (t, string) result val create_full : graph_id:Graph_types.Uuid.t -> schema:Graph_types.schema_version - -> applied_server_t:int + -> applied_server_t:Server_cursor.t -> checksum:string -> status:status -> last_error:string option diff --git a/logseq_db_worker/lib/effect_runner/effect_runner.ml b/logseq_db_worker/lib/effect_runner/effect_runner.ml index f994903..965b53d 100644 --- a/logseq_db_worker/lib/effect_runner/effect_runner.ml +++ b/logseq_db_worker/lib/effect_runner/effect_runner.ml @@ -1458,11 +1458,7 @@ let handle_sync_worker_effect t = function (match Hashtbl.find_opt t.inspections key with | None -> Error (effect_error "The snapshot mirror inspection is stale.") | Some inspection -> - let cursor = - Overlay.Server_cursor.of_string - (Printf.sprintf "server-cursor:v1:%d" request.applied_server_t) - |> Result.get_ok - in + let cursor = request.applied_server_t in (match Database.prepare_snapshot_activation t.dependencies.overlay diff --git a/logseq_db_worker/lui/logseq_db_worker_lui_service.ml b/logseq_db_worker/lui/logseq_db_worker_lui_service.ml index 0d1925c..24313a5 100644 --- a/logseq_db_worker/lui/logseq_db_worker_lui_service.ml +++ b/logseq_db_worker/lui/logseq_db_worker_lui_service.ml @@ -52,7 +52,7 @@ type snapshot = { sync_phase : sync_phase ; catalog : graph list ; selected_graph : graph_id option - ; applied_server_t : int option + ; applied_server_t : Logseq_db_types.Server_cursor.t option ; timeline_presentation_pending : bool ; startup : startup_facts ; last_error : string option diff --git a/logseq_db_worker/lui/logseq_db_worker_lui_service.mli b/logseq_db_worker/lui/logseq_db_worker_lui_service.mli index 0e7d312..df3431e 100644 --- a/logseq_db_worker/lui/logseq_db_worker_lui_service.mli +++ b/logseq_db_worker/lui/logseq_db_worker_lui_service.mli @@ -43,7 +43,7 @@ type snapshot = { sync_phase : sync_phase ; catalog : graph list ; selected_graph : graph_id option - ; applied_server_t : int option + ; applied_server_t : Logseq_db_types.Server_cursor.t option ; timeline_presentation_pending : bool ; startup : startup_facts ; last_error : string option diff --git a/logseq_db_worker/test/test_managed_sync_e2e.ml b/logseq_db_worker/test/test_managed_sync_e2e.ml index 8a136f7..2adf35b 100644 --- a/logseq_db_worker/test/test_managed_sync_e2e.ml +++ b/logseq_db_worker/test/test_managed_sync_e2e.ml @@ -1,3 +1,12 @@ +let advance_server_cursor cursor count = + let module Cursor = Logseq_db_types.Server_cursor in + let value = Cursor.to_int64 cursor in + let count = Int64.of_int count in + if count < 0L || value > Int64.sub Int64.max_int count + then invalid_arg "test cursor interval overflows"; + Cursor.of_int64 (Int64.add value count) |> Result.get_ok +;; + module Service = Logseq_db_worker_lui.Logseq_db_worker_lui_service module Core = Service module Worker = Logseq_db_worker_lui.Journal_worker @@ -185,7 +194,11 @@ let fail_worker context phase reason = | None -> "unknown", "none", "none", "none", "none" | Some state -> ( sync_phase_name state.snapshot.sync_phase - , Option.fold ~none:"none" ~some:string_of_int state.snapshot.applied_server_t + , Option.fold + ~none:"none" + ~some:(fun cursor -> + Int64.to_string (Logseq_db_types.Server_cursor.to_int64 cursor)) + state.snapshot.applied_server_t , startup_failure_stage_name state.snapshot.startup.failure , sync_error_kind state.snapshot.last_error , Option.fold ~none:"none" ~some:String.escaped state.snapshot.last_error ) @@ -214,7 +227,11 @@ let handle_public_state context state = Printf.sprintf "%s:%s:%s" (sync_phase_name snapshot.sync_phase) - (Option.fold ~none:"none" ~some:string_of_int snapshot.applied_server_t) + (Option.fold + ~none:"none" + ~some:(fun cursor -> + Int64.to_string (Logseq_db_types.Server_cursor.to_int64 cursor)) + snapshot.applied_server_t) (sync_error_kind snapshot.last_error) in if not (List.mem marker context.state_history) @@ -362,7 +379,8 @@ let await_authoritative_current context ~after_server_t phase = match context.last_state with | Some state -> (match state.snapshot.sync_phase, state.snapshot.applied_server_t with - | Current, Some cursor when cursor > after_server_t -> cursor + | Current, Some cursor + when Logseq_db_types.Server_cursor.compare cursor after_server_t > 0 -> cursor | _ -> Unix.sleepf 0.01; loop ()) @@ -768,7 +786,11 @@ let run_deployed_protocol_contract "hello-second"; let first_t = probe_hello clock first in let second_t = probe_hello clock second in - let baseline = Int.max first_t second_t in + let baseline = + if Logseq_db_types.Server_cursor.compare first_t second_t >= 0 + then first_t + else second_t + in let _, _, source_history = probe_pull clock first ~since:source_cursor in let protected_title = match @@ -802,19 +824,27 @@ let run_deployed_protocol_contract in probe_send first group "group-submit"; let group_t, _ = probe_expect_accepted clock first "group-submit" in - if group_t <> baseline + 2 + if + not + (Logseq_db_types.Server_cursor.equal + group_t + (advance_server_cursor baseline 2)) then fail "deployed group cursors are not contiguous in member order"; let pulled_t, _, pulled = probe_pull clock second ~since:baseline in - if pulled_t < group_t + if Logseq_db_types.Server_cursor.compare pulled_t group_t < 0 then fail "deployed pull did not reach the accepted group"; let group_members = pulled |> List.filter (fun transaction -> - transaction.Sync_protocol.Server.t > baseline && transaction.t <= group_t) + Logseq_db_types.Server_cursor.compare + transaction.Sync_protocol.Server.t + baseline + > 0 + && Logseq_db_types.Server_cursor.compare transaction.t group_t <= 0) in if List.map (fun tx -> tx.Sync_protocol.Server.t) group_members - <> [ baseline + 1; baseline + 2 ] + <> [ advance_server_cursor baseline 1; advance_server_cursor baseline 2 ] then fail "deployed group interval was interleaved or reordered"; List.iter2 (fun (expected_uuid, expected) transaction -> @@ -831,7 +861,7 @@ let run_deployed_protocol_contract probe_send first group "stale-retry"; (match probe_batch_result clock first with | `Rejected { reason = Sync_protocol.Stale; t = Some retry_t; _ } - when retry_t = group_t -> () + when Logseq_db_types.Server_cursor.equal retry_t group_t -> () | `Rejected _ -> fail "deployed retry did not return the current stale cursor" | `Accepted _ -> fail "deployed retry unexpectedly executed twice"); let accepted_delete_id = fresh_uuid () in @@ -858,7 +888,7 @@ let run_deployed_protocol_contract let later_remote_t, _ = probe_expect_accepted clock second "remote-after-delete" in - if delete_t >= later_remote_t + if Logseq_db_types.Server_cursor.compare delete_t later_remote_t >= 0 then fail "deployed accepted delete did not precede the remote transaction"; let stale_block = fresh_uuid () in let create_stale_id = fresh_uuid () in @@ -905,7 +935,8 @@ let run_deployed_protocol_contract ; failed_tx_id = None ; _ } - when rejection_t >= intervening_t -> () + when Logseq_db_types.Server_cursor.compare rejection_t intervening_t >= 0 -> + () | `Rejected _ -> fail "deployed stale delete lacked batch-level non-execution evidence" | `Accepted _ -> fail "deployed stale t_before delete executed"); @@ -926,7 +957,7 @@ let run_deployed_protocol_contract "second-group-marker-cleanup"; ignore (probe_expect_accepted clock first "second-group-marker-cleanup" - : int * string option)))); + : Sync_protocol.cursor * string option)))); report_phase "protocol-contract-complete" ;; @@ -1006,7 +1037,7 @@ let with_client ~label ~support ~credentials ~cognito f = ;; type durable_facts = - { cursor : int + { cursor : Logseq_db_types.Server_cursor.t ; checksum : string ; outbox_empty : bool } @@ -1043,7 +1074,8 @@ let durable_facts support graph_id = ;; let require_durable_advance facts initial_cursor label = - if facts.cursor <= initial_cursor then fail "%s durable cursor did not advance" label; + if Logseq_db_types.Server_cursor.compare facts.cursor initial_cursor <= 0 + then fail "%s durable cursor did not advance" label; if String.equal facts.checksum "0000000000000000" then fail "%s durable checksum did not advance" label; if not facts.outbox_empty then fail "%s durable outbox is not empty" label @@ -1078,7 +1110,8 @@ let best_effort_cleanup if Logseq_db_types.Graph_types.Uuid.equal graph.graph_id graph_id then ignore - (cleanup_block context ~block_uuid ~parent_uuid ~title ~opening_cursor : int)); + (cleanup_block context ~block_uuid ~parent_uuid ~title ~opening_cursor + : Logseq_db_types.Server_cursor.t)); report_phase "best-effort-cleanup-complete" with | _ -> report_phase "best-effort-cleanup-failed" @@ -1255,7 +1288,11 @@ let run () = report_phase "sender-closed"; let sender_facts = durable_facts sender_support sender_graph.graph_id in require_durable_advance sender_facts sender_opening_cursor "sender"; - if sender_facts.cursor < sender_submitted_cursor + if + Logseq_db_types.Server_cursor.compare + sender_facts.cursor + sender_submitted_cursor + < 0 then fail "sender durable cursor is behind public state"; let receiver_opening_cursor, receiver_final_cursor, receiver_graph = report_phase "receiver-start"; @@ -1289,11 +1326,19 @@ let run () = opening_cursor, final_cursor, graph) in report_phase "receiver-closed"; - if receiver_opening_cursor < sender_facts.cursor + if + Logseq_db_types.Server_cursor.compare + receiver_opening_cursor + sender_facts.cursor + < 0 then fail "receiver did not recover the sender cursor"; let receiver_facts = durable_facts receiver_support receiver_graph.graph_id in require_durable_advance receiver_facts receiver_opening_cursor "receiver"; - if receiver_facts.cursor < receiver_final_cursor + if + Logseq_db_types.Server_cursor.compare + receiver_facts.cursor + receiver_final_cursor + < 0 then fail "receiver durable cursor is behind public state"; run_deployed_protocol_contract ~token:cognito.id_token diff --git a/logseq_db_worker/test/test_pure_reducer.ml b/logseq_db_worker/test/test_pure_reducer.ml index 28bba3e..f2772bd 100644 --- a/logseq_db_worker/test/test_pure_reducer.ml +++ b/logseq_db_worker/test/test_pure_reducer.ml @@ -168,7 +168,8 @@ let worker_open_graph_transition () = |> Result.get_ok in let checkpoint = - Logseq_overlay_db.Types.Server_cursor.of_string "server-cursor:v1:0" + Logseq_overlay_db.Types.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> Result.get_ok in let sync = diff --git a/logseq_overlay_db/lib/database.ml b/logseq_overlay_db/lib/database.ml index 4c41d08..b60af8b 100644 --- a/logseq_overlay_db/lib/database.ml +++ b/logseq_overlay_db/lib/database.ml @@ -131,7 +131,7 @@ type t = ; mutable sync_revision : int ; mutable durable_sync_revision : int ; mutable durable_outbox : Logseq_db_storage.Sync_outbox_store.row Uuid_map.t - ; mutable checkpoint : int + ; mutable checkpoint : Types.server_cursor ; mutable checkpoint_metadata : Logseq_db_types.Sync_checkpoint.t ; mutable subscriptions : subscription list ; mutable snapshots : snapshot list @@ -280,10 +280,6 @@ let absent_mirror_generation path = (Digestif.SHA256.digest_string ("absent:" ^ path) |> Digestif.SHA256.to_hex) ;; -let server_cursor value = - token Types.Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" value) -;; - let dependencies ~epoch_ms ~monotonic_ns ~limits = let values = [ "response_budget_bytes", limits.response_budget_bytes @@ -346,7 +342,7 @@ let inspect_durable_mirror (location : durable_mirror_location) = Types.Available { generation = mirror_generation_for_path location.database_path ; graph_uuid = checkpoint.graph_id - ; checkpoint = server_cursor checkpoint.applied_server_t + ; checkpoint = checkpoint.applied_server_t ; checksum = Types.Checksum.of_string ("checksum:v1:" ^ checkpoint.checksum) |> Result.to_option @@ -386,11 +382,16 @@ let checksum_value checksum = | _ -> value ;; -let cursor_value cursor = - let value = Types.Server_cursor.to_string cursor in - match String.split_on_char ':' value with - | [ "server-cursor"; "v1"; ordinal ] -> int_of_string ordinal - | _ -> invalid_arg "invalid server cursor" +(* Submission ordinals are small local offsets, not server cursors. Keep the + checked interval arithmetic here; authoritative progress never narrows to int. *) +let advance_cursor cursor count = + let value = Types.Server_cursor.to_int64 cursor in + let count = Int64.of_int count in + if count < 0L || value > Int64.sub Int64.max_int count + then Error "server cursor interval overflows int64" + else + Types.Server_cursor.of_int64 (Int64.add value count) + |> Result.map_error (fun `Negative_cursor -> "negative server cursor interval") ;; let protected_snapshot_datoms database = @@ -621,7 +622,7 @@ let persist_snapshot_activation preparation = Logseq_db_types.Sync_checkpoint.create ~graph_id:preparation.snapshot_inspection.location.graph_id ~schema:(Mirror.schema preparation.snapshot_staging) - ~applied_server_t:(cursor_value preparation.snapshot_cursor) + ~applied_server_t:preparation.snapshot_cursor ~checksum:digest |> Result.map_error (fun _message -> Types.Snapshot_state_error) in @@ -2010,7 +2011,7 @@ let inspect_admission database = + Option.fold ~none:0 ~some:(fun cursor -> - String.length (Types.Server_cursor.to_string cursor)) + String.length (Persistence_json.server_cursor_to_string cursor)) record.observed_origin_cursor) 0 database.outbox @@ -4260,14 +4261,15 @@ let outbox_record_is_canonical (record : outbox_record) = record.submission_t_before, record.submission_ordinal, record.submission_count with | None, None, None -> Some None - | Some t_before, Some ordinal, Some count when count > 0 && ordinal < count -> - Some (Some (cursor_value t_before + ordinal + 1)) + | Some t_before, Some ordinal, Some count + when count > 0 && ordinal >= 0 && ordinal < count -> + Result.to_option (advance_cursor t_before (ordinal + 1)) |> Option.map Option.some | _ -> None in let origin_is_canonical = match submission_shape, record.observed_origin_cursor with | Some None, None -> true - | Some (Some expected), Some observed -> cursor_value observed = expected + | Some (Some expected), Some observed -> Types.Server_cursor.equal observed expected | Some (Some _), None -> true | None, _ | Some None, Some _ -> false in @@ -5143,13 +5145,8 @@ let open_owned ~sw dependencies inspection ~graph_name ownership = in let checkpoint = match inspection.presence with - | Types.Available { checkpoint; _ } -> - Types.Server_cursor.to_string checkpoint - |> String.split_on_char ':' - |> List.rev - |> List.hd - |> int_of_string - | Absent _ -> 0 + | Types.Available { checkpoint; _ } -> checkpoint + | Absent _ -> Types.Server_cursor.zero in let ( queryable_outbox , queryable_block_effects @@ -5648,7 +5645,7 @@ let inspect_sync database = Ok (Types.sync_view ~token:(sync_token database.sync_revision) - ~checkpoint:(server_cursor database.checkpoint) + ~checkpoint:database.checkpoint ~submissions:(List.map descriptor database.outbox))) ;; @@ -5770,20 +5767,12 @@ let rejection_suffix_records records (partition : Types.rejection_member_partiti |> fun (dependent, independent) -> List.rev dependent, List.rev independent ;; -let server_cursor_number cursor = - Types.Server_cursor.to_string cursor - |> String.split_on_char ':' - |> List.rev - |> List.hd - |> int_of_string -;; - let expected_origin_cursor (record : outbox_record) = match record.submission_t_before, record.submission_ordinal, record.submission_count with | Some t_before, Some ordinal, Some count when ordinal >= 0 && ordinal < count -> - Ok (server_cursor_number t_before + ordinal + 1) + advance_cursor t_before (ordinal + 1) | _ -> Error "own transaction omitted its frozen submission interval" ;; @@ -5797,7 +5786,7 @@ let strip_versioned_prefix prefix value = ;; let acceptance_barrier_is_current database (barrier : Types.acceptance_barrier) = - server_cursor_number barrier.through = database.checkpoint + Types.Server_cursor.equal barrier.through database.checkpoint && String.equal (strip_versioned_prefix "checksum" (Types.Checksum.to_string barrier.checksum)) (strip_versioned_prefix "checksum" database.checkpoint_metadata.checksum) @@ -5982,6 +5971,11 @@ let begin_outbox_transition database ~expected transition = (fun (record : outbox_record) -> record.transport_state <> Types.Queued) records then Error (Types.Outbox_transition_invalid "submit member is not queued") + else if + Result.is_error (advance_cursor database.checkpoint (List.length records)) + then + Error + (Types.Outbox_transition_invalid "server cursor interval overflows int64") else if not (submission_records_are_in_sequence_order records) then Error (Types.Outbox_transition_invalid "submit members are reordered") else if Option.is_some ineligible_dependency @@ -6059,9 +6053,9 @@ let begin_outbox_transition database ~expected transition = -> (match executed_through with | Some executed - when server_cursor_number executed <= server_cursor_number barrier.through - && server_cursor_number barrier.through <= server_cursor_number through - -> make Duplicate_plan None + when Types.Server_cursor.compare executed barrier.through <= 0 + && Types.Server_cursor.compare barrier.through through <= 0 -> + make Duplicate_plan None | _ -> Error (Types.Outbox_transition_invalid @@ -6079,16 +6073,16 @@ let begin_outbox_transition database ~expected transition = Error (Types.Outbox_transition_invalid "accept batch is missing") | Ok None -> let records = batch_records database id in - let through = server_cursor_number barrier.through in + let through = barrier.through in let validate result (record : outbox_record) = Result.bind result (fun () -> Result.bind (expected_origin_cursor record) (fun expected -> - if expected > through + if Types.Server_cursor.compare expected through > 0 then Error "acceptance barrier precedes its own submission interval" - else if through <= database.checkpoint + else if Types.Server_cursor.compare through database.checkpoint <= 0 then ( match record.observed_origin_cursor with - | Some cursor when server_cursor_number cursor = expected -> Ok () + | Some cursor when Types.Server_cursor.equal cursor expected -> Ok () | None | Some _ -> Error "covered acceptance has no durable origin evidence") else Ok ())) @@ -6097,7 +6091,7 @@ let begin_outbox_transition database ~expected transition = | Error message -> Error (Types.Outbox_transition_invalid message) | Ok () -> if - through = database.checkpoint + Types.Server_cursor.equal through database.checkpoint && not (acceptance_barrier_is_current database barrier) then Error (Types.Outbox_transition_invalid "acceptance checksum mismatch") else make (Accept_plan (id, barrier)) None)) @@ -6125,8 +6119,7 @@ let begin_outbox_transition database ~expected transition = List.exists (fun (record : outbox_record) -> match record.submission_t_before with - | Some baseline -> - server_cursor_number through <= server_cursor_number baseline + | Some baseline -> Types.Server_cursor.compare through baseline <= 0 | None -> true) records then @@ -6293,7 +6286,7 @@ let make_batch preparation encrypted = | Retry_plan (_, record :: _) -> Option.get record.submission_t_before | Retry_plan (_, []) -> invalid_arg "retry batch is empty" | Submit_plan _ | Reject_plan _ | Accept_plan _ | Duplicate_plan -> - server_cursor preparation.transition_owner.checkpoint) + preparation.transition_owner.checkpoint) ~wires |> Result.get_ok ;; @@ -6346,50 +6339,69 @@ let outbox_fits_submission preparation records batch = ;; let outbox_submission_batch preparation ~encrypted = - match preparation.plan, preparation.protection_request, encrypted with - | Submit_plan _, Some request, Some (actual_request, encrypted) - when actual_request == request -> - (match validate_protected_values ~request ~encrypted with - | Error error -> Error (Types.Outbox_crypto_error error) - | Ok () -> - let batch = make_batch preparation (Some encrypted) in + let interval = + match preparation.plan with + | Submit_plan records -> + advance_cursor preparation.transition_owner.checkpoint (List.length records) + |> Result.map (fun _ -> ()) + | Retry_plan (_, (record :: _ as records)) -> + advance_cursor (Option.get record.submission_t_before) (List.length records) + |> Result.map (fun _ -> ()) + | Reject_plan (id, Definitive { partition; _ }) -> + let records = batch_records preparation.transition_owner id in + let _, independent = rejection_suffix_records records partition in + advance_cursor preparation.transition_owner.checkpoint (List.length independent) + |> Result.map (fun _ -> ()) + | Retry_plan (_, []) | Accept_plan _ | Reject_plan (_, Stale _) | Duplicate_plan -> + Ok () + in + match interval with + | Error message -> Error (Types.Outbox_transition_invalid message) + | Ok () -> + (match preparation.plan, preparation.protection_request, encrypted with + | Submit_plan _, Some request, Some (actual_request, encrypted) + when actual_request == request -> + (match validate_protected_values ~request ~encrypted with + | Error error -> Error (Types.Outbox_crypto_error error) + | Ok () -> + let batch = make_batch preparation (Some encrypted) in + let records = + match preparation.plan with + | Submit_plan records -> records + | _ -> assert false + in + if not (outbox_fits_submission preparation records batch) + then Error Types.Outbox_limit_exceeded + else Ok (Some batch)) + | Submit_plan _, Some _, None -> Error Types.Outbox_crypto_required + | Submit_plan _, Some _, Some _ -> Error Types.Outbox_crypto_unexpected + | Retry_plan _, None, None -> + let batch = make_batch preparation None in let records = match preparation.plan with - | Submit_plan records -> records + | Retry_plan (_, records) -> records | _ -> assert false in if not (outbox_fits_submission preparation records batch) then Error Types.Outbox_limit_exceeded - else Ok (Some batch)) - | Submit_plan _, Some _, None -> Error Types.Outbox_crypto_required - | Submit_plan _, Some _, Some _ -> Error Types.Outbox_crypto_unexpected - | Retry_plan _, None, None -> - let batch = make_batch preparation None in - let records = - match preparation.plan with - | Retry_plan (_, records) -> records - | _ -> assert false - in - if not (outbox_fits_submission preparation records batch) - then Error Types.Outbox_limit_exceeded - else Ok (Some batch) - | Reject_plan (_, Definitive { partition; _ }), None, None -> - let id = - match preparation.plan with - | Reject_plan (id, _) -> id - | _ -> assert false - in - let records = batch_records preparation.transition_owner id in - let _dependent, independent = rejection_suffix_records records partition in - Ok - (match independent with - | [] -> None - | _ -> Some (make_batch preparation None)) - | (Accept_plan _ | Reject_plan _ | Duplicate_plan), None, None -> Ok None - | (Retry_plan _ | Accept_plan _ | Reject_plan _ | Duplicate_plan), _, Some _ -> - Error Types.Outbox_crypto_unexpected - | _ -> - Error (Types.Outbox_transition_invalid "preparation crypto state is inconsistent") + else Ok (Some batch) + | Reject_plan (_, Definitive { partition; _ }), None, None -> + let id = + match preparation.plan with + | Reject_plan (id, _) -> id + | _ -> assert false + in + let records = batch_records preparation.transition_owner id in + let _dependent, independent = rejection_suffix_records records partition in + Ok + (match independent with + | [] -> None + | _ -> Some (make_batch preparation None)) + | (Accept_plan _ | Reject_plan _ | Duplicate_plan), None, None -> Ok None + | (Retry_plan _ | Accept_plan _ | Reject_plan _ | Duplicate_plan), _, Some _ -> + Error Types.Outbox_crypto_unexpected + | _ -> + Error (Types.Outbox_transition_invalid "preparation crypto state is inconsistent")) ;; let update_submitted database ~capture_dependency_shadows records batch = @@ -6434,12 +6446,15 @@ let begin_authoritative database ~expected batch = else ( let transactions = Types.authoritative_batch_transactions batch in let cursors = List.map Types.authoritative_transaction_cursor transactions in - let rec continuous expected = function + let rec continuous previous = function | [] -> true | cursor :: rest -> - server_cursor_number cursor = expected && continuous (expected + 1) rest + (match advance_cursor previous 1 with + | Error _ -> false + | Ok expected -> + Types.Server_cursor.equal cursor expected && continuous cursor rest) in - if not (continuous (database.checkpoint + 1) cursors) + if not (continuous database.checkpoint cursors) then Error Types.Authoritative_cursor_discontinuous else ( let wires = @@ -6533,7 +6548,7 @@ let transaction_matches_record_at_cursor record cursor root_before transaction_d match expected_origin_cursor record with | Error _ -> false | Ok expected_cursor -> - if server_cursor_number cursor <> expected_cursor + if not (Types.Server_cursor.equal cursor expected_cursor) then false else ( match expected_record_transaction_data root_before record with @@ -6603,39 +6618,39 @@ let validate_active_transaction_origins owner batch transaction_data roots_after Result.bind validated (fun reversed -> let validated = List.rev reversed in let previous_checkpoint = owner.checkpoint in - let through = server_cursor_number (Types.authoritative_batch_through batch) in + let through = Types.authoritative_batch_through batch in let validate_accepted result (record : outbox_record) = Result.bind result (fun () -> match record.transport_state with | Types.Accepted_pending_authoritative _ -> Result.bind (expected_origin_cursor record) (fun expected -> - if expected <= previous_checkpoint + if Types.Server_cursor.compare expected previous_checkpoint <= 0 then ( match record.observed_origin_cursor with - | Some cursor when server_cursor_number cursor = expected -> Ok () + | Some cursor when Types.Server_cursor.equal cursor expected -> Ok () | None | Some _ -> Error "accepted transaction has no durable origin evidence") - else if expected <= through + else if Types.Server_cursor.compare expected through <= 0 then if List.exists (fun (origin, cursor) -> Graph.Uuid.equal origin record.mutation_id - && server_cursor_number cursor = expected) + && Types.Server_cursor.equal cursor expected) validated then Ok () else ( let actual = contexts |> List.find_opt (fun (cursor, _, _) -> - server_cursor_number cursor = expected) + Types.Server_cursor.equal cursor expected) |> Option.map (fun (_, data, _) -> transaction_data_shape data) |> Option.value ~default:"missing-cursor" in let expected_shape = contexts |> List.find_opt (fun (cursor, _, _) -> - server_cursor_number cursor = expected) + Types.Server_cursor.equal cursor expected) |> fun context -> Option.bind context (fun (_, _, root_before) -> expected_record_transaction_data root_before record @@ -6853,7 +6868,7 @@ let validated_record_origin (record : outbox_record) origins = | None -> record.observed_origin_cursor in match expected_origin_cursor record, observed with - | Ok expected, Some cursor when expected = server_cursor_number cursor -> Some cursor + | Ok expected, Some cursor when Types.Server_cursor.equal expected cursor -> Some cursor | _ -> None ;; @@ -6891,7 +6906,7 @@ let classify_stale_deletes owner batch transactions roots_after = | Delete_blocks _, None -> Ok (`Blocked (record.mutation_id, batch_id)) | _ -> Error "stale delete state contains a non-delete mutation")) | cursor :: cursor_rest, _operations :: operation_rest, after :: root_rest -> - if server_cursor_number cursor > server_cursor_number rejection_through + if Types.Server_cursor.compare cursor rejection_through > 0 then classify_record record @@ -7121,7 +7136,7 @@ let prepare_authoritative_candidate preparation ~decrypted = (match List.find_opt (fun (cursor, _, _, _) -> - server_cursor_number cursor > server_cursor_number t_before) + Types.Server_cursor.compare cursor t_before > 0) contexts with | None -> Ok (`Deferred batch_id) @@ -7552,7 +7567,7 @@ let settle_stale_records ~checkpoint ~origins records = let covered (record : outbox_record) = match record.transport_state with | Stale_rejected_pending_authoritative { through; _ } -> - server_cursor_number through <= checkpoint + Types.Server_cursor.compare through checkpoint <= 0 | _ -> false in let settled = @@ -7639,7 +7654,7 @@ let settle_stale_records ~checkpoint ~origins records = | None, cursor -> cursor | Some left, Some right -> Some - (if server_cursor_number left > server_cursor_number right + (if Types.Server_cursor.compare left right > 0 then left else right) | cursor, None -> cursor) @@ -7721,11 +7736,11 @@ let apply_outbox_transition database preparation ~encrypted = let logical_before = match preparation.plan with | Accept_plan (_, barrier) - when server_cursor_number barrier.through <= database.checkpoint -> - Some (snapshot_of_database database) + when Types.Server_cursor.compare barrier.through database.checkpoint + <= 0 -> Some (snapshot_of_database database) | Reject_plan (_, Definitive _) -> Some (snapshot_of_database database) | Reject_plan (_, Stale { through }) - when server_cursor_number through <= database.checkpoint -> + when Types.Server_cursor.compare through database.checkpoint <= 0 -> Some (snapshot_of_database database) | Submit_plan _ | Retry_plan _ | Accept_plan _ | Reject_plan (_, Stale _) @@ -7773,7 +7788,8 @@ let apply_outbox_transition database preparation ~encrypted = records; Types.Logically_active, [] | Accept_plan (id, barrier) - when server_cursor_number barrier.through > database.checkpoint -> + when Types.Server_cursor.compare barrier.through database.checkpoint > 0 + -> List.iter (fun (record : outbox_record) -> record.transport_state <- Types.Accepted_pending_authoritative id; @@ -7784,7 +7800,7 @@ let apply_outbox_transition database preparation ~encrypted = let records = batch_records database id in let current_root = authoritative_database database in let barrier_is_behind = - server_cursor_number barrier.through < database.checkpoint + Types.Server_cursor.compare barrier.through database.checkpoint < 0 in let incorporated, mismatched = List.partition @@ -8471,7 +8487,6 @@ let commit_authoritative_candidate database prepared = | Ok staged -> let through = Types.authoritative_batch_through preparation.authoritative_batch - |> server_cursor_number in let authoritative_after = prepared.authoritative_db_after in let authoritative_before = authoritative_database database in @@ -8533,13 +8548,13 @@ let commit_authoritative_candidate database prepared = (fun (record : outbox_record) -> match record.transport_state, record.acceptance_barrier with | Accepted_pending_authoritative _, Some barrier -> - server_cursor_number barrier.through <= through + Types.Server_cursor.compare barrier.through through <= 0 | _ -> false) database.outbox in let authoritative_root_at barrier = - let target = server_cursor_number barrier.Types.through in - if target <= database.checkpoint + let target = barrier.Types.through in + if Types.Server_cursor.compare target database.checkpoint <= 0 then Some authoritative_before else List.combine @@ -8547,9 +8562,9 @@ let commit_authoritative_candidate database prepared = prepared.authoritative_roots_after |> List.find_map (fun (transaction, root) -> if - server_cursor_number + Types.Server_cursor.equal (Types.authoritative_transaction_cursor transaction) - = target + target then Some root else None) in @@ -8557,14 +8572,15 @@ let commit_authoritative_candidate database prepared = match record.acceptance_barrier with | None -> false | Some barrier -> - let barrier_number = server_cursor_number barrier.through in + let barrier_cursor = barrier.through in (match record.mutation with | Save_block _ | Insert_blocks _ | Create_journal_page _ | Set_task_status _ | Clear_task_status _ - when barrier_number < database.checkpoint -> true + when Types.Server_cursor.compare barrier_cursor database.checkpoint < 0 + -> true | _ -> (match authoritative_root_at barrier, record.transport_state with | Some root, Accepted_pending_authoritative batch_id -> @@ -8989,7 +9005,7 @@ let commit_authoritative_candidate database prepared = { Types.generation = database.generation ; before_projection_revision = before_revision ; after_projection_revision = after_revision - ; checkpoint = server_cursor through + ; checkpoint = through ; sync_token = sync_token database.sync_revision ; terminal_receipts ; replanned_queued_ids = diff --git a/logseq_overlay_db/lib/persistence_json.ml b/logseq_overlay_db/lib/persistence_json.ml index c6949e6..db2b2ab 100644 --- a/logseq_overlay_db/lib/persistence_json.ml +++ b/logseq_overlay_db/lib/persistence_json.ml @@ -63,3 +63,30 @@ let block_reason_of_yojson = function | `String "plannerDependencyChanged" -> Ok Types.Planner_dependency_changed | _ -> Error "invalid block reason" ;; + +let server_cursor_to_string cursor = + "server-cursor:v1:" ^ Int64.to_string (Types.Server_cursor.to_int64 cursor) +;; + +let server_cursor_of_string value = + let prefix = "server-cursor:v1:" in + let length = String.length prefix in + if String.length value <= length || String.sub value 0 length <> prefix + then Error "invalid server cursor token version" + else ( + let digits = String.sub value length (String.length value - length) in + if + not + (String.for_all + (function + | '0' .. '9' -> true + | _ -> false) + digits) + then Error "invalid server cursor number" + else ( + match Int64.of_string_opt digits with + | None -> Error "server cursor overflows int64" + | Some value -> + Types.Server_cursor.of_int64 value + |> Result.map_error (fun `Negative_cursor -> "negative server cursor"))) +;; diff --git a/logseq_overlay_db/lib/persistence_outbox_v16.ml b/logseq_overlay_db/lib/persistence_outbox_v16.ml index ffa857e..3654d7b 100644 --- a/logseq_overlay_db/lib/persistence_outbox_v16.ml +++ b/logseq_overlay_db/lib/persistence_outbox_v16.ml @@ -599,7 +599,7 @@ let transport_state_to_yojson = function | Stale_rejected_pending_authoritative { batch_id; through } -> `Assoc [ "batchId", `String (Types.Submission_batch_id.to_string batch_id) - ; "through", `String (Types.Server_cursor.to_string through) + ; "through", `String (Persistence_json.server_cursor_to_string through) ; "type", `String "staleRejectedPendingAuthoritative" ] | Blocked -> `Assoc [ "type", `String "blocked" ] @@ -620,7 +620,7 @@ let transport_state_of_yojson = function ; ("type", `String "staleRejectedPendingAuthoritative") ] -> let* batch_id = Types.Submission_batch_id.of_string id in - let* through = Types.Server_cursor.of_string through in + let* through = Persistence_json.server_cursor_of_string through in Ok (Types.Stale_rejected_pending_authoritative { batch_id; through }) | `Assoc [ ("type", `String "blocked") ] -> Ok Types.Blocked | _ -> Error "invalid transport state" @@ -639,13 +639,13 @@ type acceptance_barrier_v16 = let acceptance_barrier_to_dto (barrier : Types.acceptance_barrier) = { checksum = Types.Checksum.to_string barrier.checksum - ; through = Types.Server_cursor.to_string barrier.through + ; through = Persistence_json.server_cursor_to_string barrier.through } ;; let acceptance_barrier_of_dto barrier = let* checksum = Types.Checksum.of_string barrier.checksum in - let* through = Types.Server_cursor.of_string barrier.through in + let* through = Persistence_json.server_cursor_of_string barrier.through in Ok Types.{ through; checksum } ;; @@ -719,13 +719,15 @@ let encode ~sync_revision (record : t) = ; state = record.transport_state ; same_id_retry_eligible = record.same_id_retry_eligible ; submission_t_before = - Option.map Types.Server_cursor.to_string record.submission_t_before + Option.map Persistence_json.server_cursor_to_string record.submission_t_before ; submission_ordinal = record.submission_ordinal ; submission_count = record.submission_count ; observed_origin_cursor = - Option.map Types.Server_cursor.to_string record.observed_origin_cursor + Option.map Persistence_json.server_cursor_to_string record.observed_origin_cursor ; stale_earliest_conflict_cursor = - Option.map Types.Server_cursor.to_string record.stale_earliest_conflict_cursor + Option.map + Persistence_json.server_cursor_to_string + record.stale_earliest_conflict_cursor ; stale_conflicts = List.map Delete_conflict.code record.stale_conflicts ; sync_revision } @@ -813,13 +815,15 @@ let decode source = | Some value -> Result.map Option.some (acceptance_barrier_of_dto value) in let* submission_t_before = - optional_token Types.Server_cursor.of_string dto.submission_t_before + optional_token Persistence_json.server_cursor_of_string dto.submission_t_before in let* observed_origin_cursor = - optional_token Types.Server_cursor.of_string dto.observed_origin_cursor + optional_token Persistence_json.server_cursor_of_string dto.observed_origin_cursor in let* stale_earliest_conflict_cursor = - optional_token Types.Server_cursor.of_string dto.stale_earliest_conflict_cursor + optional_token + Persistence_json.server_cursor_of_string + dto.stale_earliest_conflict_cursor in let* stale_conflicts = decode_conflicts dto.stale_conflicts in Ok diff --git a/logseq_overlay_db/lib/persistence_receipt_v1.ml b/logseq_overlay_db/lib/persistence_receipt_v1.ml index d358ee6..238b467 100644 --- a/logseq_overlay_db/lib/persistence_receipt_v1.ml +++ b/logseq_overlay_db/lib/persistence_receipt_v1.ml @@ -100,7 +100,7 @@ let encode_mutation (mutation_id, fingerprint, entry) = Types.delete_conflict_kinds receipt.conflicts |> List.map Delete_conflict.code ; earliest_conflict_cursor = Option.map - Types.Server_cursor.to_string + Persistence_json.server_cursor_to_string (Types.remote_won_proof_earliest_conflict_cursor proof) ; fingerprint ; format_version = 1 @@ -113,11 +113,11 @@ let encode_mutation (mutation_id, fingerprint, entry) = | Proven_unexecuted -> "provenUnexecuted") ; rejection_through = Option.map - Types.Server_cursor.to_string + Persistence_json.server_cursor_to_string (Types.remote_won_proof_rejection_through proof) ; t_before = Option.map - Types.Server_cursor.to_string + Persistence_json.server_cursor_to_string (Types.remote_won_proof_t_before proof) } in @@ -218,12 +218,16 @@ let decode_remote_won ~key json = let* batch_id = optional_token Types.Submission_batch_id.of_string receipt.batch_id in - let* t_before = optional_token Types.Server_cursor.of_string receipt.t_before in + let* t_before = + optional_token Persistence_json.server_cursor_of_string receipt.t_before + in let* rejection_through = - optional_token Types.Server_cursor.of_string receipt.rejection_through + optional_token Persistence_json.server_cursor_of_string receipt.rejection_through in let* earliest_conflict_cursor = - optional_token Types.Server_cursor.of_string receipt.earliest_conflict_cursor + optional_token + Persistence_json.server_cursor_of_string + receipt.earliest_conflict_cursor in let* proof = Types.remote_won_proof @@ -294,8 +298,9 @@ let encode_terminal_batch receipt = stale_terminal_batch_receipt_v1_to_yojson { batch_id = Types.Submission_batch_id.to_string receipt.terminal_batch_id ; format_version = 1 - ; through = Types.Server_cursor.to_string through - ; executed_through = Option.map Types.Server_cursor.to_string executed_through + ; through = Persistence_json.server_cursor_to_string through + ; executed_through = + Option.map Persistence_json.server_cursor_to_string executed_through ; receipt_type = "staleBatch" } | Terminal_accepted barrier -> @@ -304,7 +309,7 @@ let encode_terminal_batch receipt = ; format_version = 1 ; checksum = Types.Checksum.to_string barrier.checksum ; receipt_type = "acceptedBatch" - ; through = Types.Server_cursor.to_string barrier.through + ; through = Persistence_json.server_cursor_to_string barrier.through } | Terminal_proven_unexecuted { mutation_id; fingerprint } -> proven_unexecuted_terminal_batch_receipt_v1_to_yojson @@ -327,11 +332,12 @@ let decode_terminal_batch (key, source) = | Some (`String "staleBatch") -> let* receipt = stale_terminal_batch_receipt_v1_of_yojson json in let* terminal_batch_id = Types.Submission_batch_id.of_string receipt.batch_id in - let* through = Types.Server_cursor.of_string receipt.through in + let* through = Persistence_json.server_cursor_of_string receipt.through in let* executed_through = match receipt.executed_through with | None -> Ok None - | Some value -> Result.map Option.some (Types.Server_cursor.of_string value) + | Some value -> + Result.map Option.some (Persistence_json.server_cursor_of_string value) in if receipt.format_version <> 1 @@ -351,7 +357,7 @@ let decode_terminal_batch (key, source) = then Error "invalid terminal batch receipt identity or version" else let* checksum = Types.Checksum.of_string receipt.checksum in - let* through = Types.Server_cursor.of_string receipt.through in + let* through = Persistence_json.server_cursor_of_string receipt.through in Ok { terminal_batch_id ; terminal_outcome = Terminal_accepted Types.{ through; checksum } diff --git a/logseq_overlay_db/lib/server_cursor.ml b/logseq_overlay_db/lib/server_cursor.ml new file mode 100644 index 0000000..4fc3f58 --- /dev/null +++ b/logseq_overlay_db/lib/server_cursor.ml @@ -0,0 +1 @@ +include Logseq_db_types.Server_cursor diff --git a/logseq_overlay_db/lib/types.ml b/logseq_overlay_db/lib/types.ml index ad59cbc..ac4fc53 100644 --- a/logseq_overlay_db/lib/types.ml +++ b/logseq_overlay_db/lib/types.ml @@ -63,9 +63,7 @@ module Mirror_generation = Make_versioned_token (struct let prefix = "mirror-generation" end) -module Server_cursor = Make_versioned_token (struct - let prefix = "server-cursor" - end) +module Server_cursor = Server_cursor module Checksum = Make_versioned_token (struct let prefix = "checksum" diff --git a/logseq_overlay_db/spec/dune b/logseq_overlay_db/spec/dune index 5cf280d..98cfdb5 100644 --- a/logseq_overlay_db/spec/dune +++ b/logseq_overlay_db/spec/dune @@ -1,7 +1,7 @@ (library (name logseq_overlay_db) (public_name logseq_overlay_db) - (modules types database) - (virtual_modules types database) + (modules server_cursor types database) + (virtual_modules server_cursor types database) (default_implementation logseq_overlay_db_impl) (libraries eio logseq_db_types)) diff --git a/logseq_overlay_db/spec/server_cursor.mli b/logseq_overlay_db/spec/server_cursor.mli new file mode 100644 index 0000000..5b6e3f0 --- /dev/null +++ b/logseq_overlay_db/spec/server_cursor.mli @@ -0,0 +1,8 @@ +(** A non-negative authoritative server transaction cursor. *) +type t = Logseq_db_types.Server_cursor.t + +val zero : t +val of_int64 : int64 -> (t, [ `Negative_cursor ]) result +val to_int64 : t -> int64 +val equal : t -> t -> bool +val compare : t -> t -> int diff --git a/logseq_overlay_db/spec/types.mli b/logseq_overlay_db/spec/types.mli index 5ce2b3c..0e98cf8 100644 --- a/logseq_overlay_db/spec/types.mli +++ b/logseq_overlay_db/spec/types.mli @@ -20,7 +20,7 @@ module Block_state_revision : VERSIONED_TOKEN module Page_state_revision : VERSIONED_TOKEN module Scope_revision : VERSIONED_TOKEN module Mirror_generation : VERSIONED_TOKEN -module Server_cursor : VERSIONED_TOKEN +module Server_cursor = Server_cursor module Checksum : VERSIONED_TOKEN module Crypto_item_id : VERSIONED_TOKEN module Submission_batch_id : VERSIONED_TOKEN diff --git a/logseq_overlay_db/test/test_asset_reads.ml b/logseq_overlay_db/test/test_asset_reads.ml index 011aa4e..a0ea51d 100644 --- a/logseq_overlay_db/test/test_asset_reads.ml +++ b/logseq_overlay_db/test/test_asset_reads.ml @@ -17,7 +17,9 @@ let apply database number ops = |> get in let cursor = - Types.Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" number) |> get + Types.Server_cursor.of_int64 (Int64.of_int number) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> get in let batch = Types.authoritative_batch diff --git a/logseq_overlay_db/test/test_overlay_changes.ml b/logseq_overlay_db/test/test_overlay_changes.ml index 33c68c5..0394a61 100644 --- a/logseq_overlay_db/test/test_overlay_changes.ml +++ b/logseq_overlay_db/test/test_overlay_changes.ml @@ -43,7 +43,8 @@ let commit_insert database behavior ordinal = ;; let server_cursor ordinal = - Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" ordinal) + Server_cursor.of_int64 (Int64.of_int ordinal) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> T.require_ok ~behavior:"construct listener server cursor" ;; @@ -520,7 +521,11 @@ let authoritative_membership_change_reports_structure_interests database = |> encoded_transaction_of_string ~maximum_bytes:4096 |> T.require_ok ~behavior in - let cursor = Server_cursor.of_string "server-cursor:v1:1" |> T.require_ok ~behavior in + let cursor = + Server_cursor.of_int64 1L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> T.require_ok ~behavior + in let batch = authoritative_batch ~maximum_count:16 diff --git a/logseq_overlay_db/test/test_overlay_mutations.ml b/logseq_overlay_db/test/test_overlay_mutations.ml index 0fbc67b..d9c7a30 100644 --- a/logseq_overlay_db/test/test_overlay_mutations.ml +++ b/logseq_overlay_db/test/test_overlay_mutations.ml @@ -774,7 +774,11 @@ let append_orders_follow_existing_tail database = ]) [ T.authoritative_block_uuid, "a0"; T.reference_source_uuid, "a1" ])) in - let cursor = Server_cursor.of_string "server-cursor:v1:1" |> T.require_ok ~behavior in + let cursor = + Server_cursor.of_int64 1L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> T.require_ok ~behavior + in let transaction = encoded_transaction_of_string ~maximum_bytes:4096 wire |> T.require_ok ~behavior in @@ -1026,7 +1030,9 @@ let orders_survive_reopen_and_outbox_drain () = |> T.require_ok ~behavior in let cursor = - Server_cursor.of_string "server-cursor:v1:1" |> T.require_ok ~behavior + Server_cursor.of_int64 1L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> T.require_ok ~behavior in let batch = authoritative_batch diff --git a/logseq_overlay_db/test/test_overlay_reads.ml b/logseq_overlay_db/test/test_overlay_reads.ml index 0226258..9a77159 100644 --- a/logseq_overlay_db/test/test_overlay_reads.ml +++ b/logseq_overlay_db/test/test_overlay_reads.ml @@ -79,7 +79,11 @@ let point_reads_preserve_property_summaries database = |> encoded_transaction_of_string ~maximum_bytes:4_096 |> T.require_ok ~behavior in - let cursor = Server_cursor.of_string "server-cursor:v1:1" |> T.require_ok ~behavior in + let cursor = + Server_cursor.of_int64 1L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> T.require_ok ~behavior + in let batch = authoritative_batch ~maximum_count:16 @@ -484,7 +488,9 @@ let verify_continuation_above_maximum_offset_fails_the_page database = |> T.require_ok ~behavior in let server_cursor = - Server_cursor.of_string "server-cursor:v1:1" |> T.require_ok ~behavior + Server_cursor.of_int64 1L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> T.require_ok ~behavior in let batch = authoritative_batch @@ -779,7 +785,11 @@ let indexed_journals_preserve_selection database = |> encoded_transaction_of_string ~maximum_bytes:262144 |> T.require_ok ~behavior in - let cursor = Server_cursor.of_string "server-cursor:v1:1" |> T.require_ok ~behavior in + let cursor = + Server_cursor.of_int64 1L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> T.require_ok ~behavior + in let batch = authoritative_batch ~maximum_count:16 @@ -918,7 +928,8 @@ let page_tree_preserves_valid_windows_and_snapshots database = |> T.require_ok ~behavior in let cursor = - Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" number) + Server_cursor.of_int64 (Int64.of_int number) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> T.require_ok ~behavior in let batch = @@ -1095,7 +1106,8 @@ let favorites_resolve_order_filter_and_snapshot database = |> T.require_ok ~behavior in let cursor = - Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" number) + Server_cursor.of_int64 (Int64.of_int number) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> T.require_ok ~behavior in let batch = diff --git a/logseq_overlay_db/test/test_overlay_storage.ml b/logseq_overlay_db/test/test_overlay_storage.ml index 070b3ee..93e8df1 100644 --- a/logseq_overlay_db/test/test_overlay_storage.ml +++ b/logseq_overlay_db/test/test_overlay_storage.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + module Database = Logseq_overlay_db.Database module T = Test_support module Types = Logseq_overlay_db.Types @@ -47,7 +51,8 @@ let page_precondition database page behavior = ;; let server_cursor value = - Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" value) + Server_cursor.of_int64 (Int64.of_int value) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> T.require_ok ~behavior:"construct server cursor" ;; @@ -1435,7 +1440,8 @@ let authoritative_checkpoint_and_root_survive_reopen () = |> T.require_ok ~behavior in let cursor = - Server_cursor.of_string "server-cursor:v1:1" |> T.require_ok ~behavior + Server_cursor.of_int64 1L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> T.require_ok ~behavior in let batch = authoritative_batch @@ -2589,4 +2595,109 @@ let cases = ] ;; -let () = Alcotest.run "logseq_overlay_db storage" [ "mirror and durability", cases ] +(* The database owns interval freezing; the reducer only requests this transition. + Exercise its public mutation and outbox boundaries without injected results. *) +let submit_interval_overflow_is_rejected () = + let behavior = "submit interval overflow is rejected before freezing" in + T.with_temp_directory "cursor-overflow-" (fun support -> + let path = T.seed_mirror support in + let sqlite = Sqlite3.db_open path in + let module Store = Logseq_db_storage.Sync_checkpoint_store in + let metadata = Store.read_database sqlite |> Result.get_ok in + let metadata = + { metadata with + applied_server_t = Server_cursor.of_int64 Int64.max_int |> Result.get_ok + } + in + Store.update_database sqlite metadata |> Result.get_ok; + T.require (Sqlite3.db_close sqlite) "checkpoint handle closes"; + Eio_posix.run (fun _ -> + Eio.Switch.run (fun sw -> + let dependencies = T.dependencies ~behavior in + let open_database () = + Database.open_ + ~sw + dependencies + (inspect_mirror support behavior) + ~graph_name:"overflow" + |> T.require_ok ~behavior + in + let database = open_database () in + Database.commit_local + database + ~expected:(T.insert_precondition database ~parent:T.page_uuid ~behavior) + (T.insert_blocks ~ordinal:200 ()) + |> T.require_ok ~behavior + |> ignore; + let view = Database.inspect_sync database |> T.require_ok ~behavior in + (match + Database.begin_outbox_transition + database + ~expected:(sync_view_token view) + (Submit_group [ T.mutation_uuid 200 ]) + with + | Error (Outbox_transition_invalid _) -> () + | _ -> Alcotest.fail "overflowing submit interval was accepted"); + Database.close database |> T.require_ok ~behavior; + let reopened = open_database () in + let view = Database.inspect_sync reopened |> T.require_ok ~behavior in + T.require + (List.for_all + (fun member -> member.state = Queued) + (sync_view_submissions view)) + "overflow rejection preserves queued outbox after reopen"; + Database.close reopened |> T.require_ok ~behavior))) +;; + +let sqlite_full_int64_checkpoint () = + let module Checkpoint = Logseq_db_types.Sync_checkpoint in + let module Store = Logseq_db_storage.Sync_checkpoint_store in + let sqlite = Sqlite3.db_open ":memory:" in + Fun.protect + ~finally:(fun () -> ignore (Sqlite3.db_close sqlite)) + (fun () -> + let metadata = + Checkpoint.create + ~graph_id:T.graph_uuid + ~schema:Logseq_db_types.Graph_types.{ major = 1; minor = 0 } + ~applied_server_t:(server_cursor_of_int 0) + ~checksum:"0000000000000000" + |> Result.get_ok + in + Store.initialize_database sqlite metadata |> Result.get_ok; + Sqlite3.exec sqlite "UPDATE sync_meta SET applied_server_t = 9223372036854775807" + |> Sqlite3.Rc.check; + let restored = Store.read_database sqlite |> Result.get_ok in + Alcotest.(check int64) + "SQLite reads the complete int64 value" + Int64.max_int + (Server_cursor.to_int64 restored.applied_server_t); + let next = + { restored with + applied_server_t = + Server_cursor.of_int64 (Int64.pred Int64.max_int) |> Result.get_ok + } + in + Store.update_database sqlite next |> Result.get_ok; + let roundtrip = Store.read_database sqlite |> Result.get_ok in + T.require + (Server_cursor.equal next.applied_server_t roundtrip.applied_server_t) + "SQLite update preserves the complete cursor") +;; + +let () = + Alcotest.run + "logseq_overlay_db storage" + [ ( "cursor checkpoint" + , [ Alcotest.test_case + "submit interval overflow is rejected" + `Quick + submit_interval_overflow_is_rejected + ; Alcotest.test_case + "SQLite full int64 checkpoint" + `Quick + sqlite_full_int64_checkpoint + ] ) + ; "mirror and durability", cases + ] +;; diff --git a/logseq_overlay_db/test/test_overlay_sync.ml b/logseq_overlay_db/test/test_overlay_sync.ml index ee729c3..f08a3ee 100644 --- a/logseq_overlay_db/test/test_overlay_sync.ml +++ b/logseq_overlay_db/test/test_overlay_sync.ml @@ -4,7 +4,8 @@ module Types = Logseq_overlay_db.Types open Types let server_cursor value = - Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" value) + Server_cursor.of_int64 (Int64.of_int value) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> T.require_ok ~behavior:"construct server cursor" ;; @@ -3718,8 +3719,32 @@ let stale_invalid_barrier_is_rejected database = | _ -> Alcotest.fail "non-advancing Stale barrier was admitted as non-execution proof" ;; +let numeric_cursor_construction_and_comparison () = + let cursor n = + Server_cursor.of_int64 n + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + in + List.iter + (fun n -> + T.require + (Result.is_error (cursor n)) + "negative authoritative cursor was constructible") + [ -1L; Int64.min_int ]; + List.iter + (fun (a, b) -> + let a = cursor a |> T.require_ok ~behavior:"numeric cursor" in + let b = cursor b |> T.require_ok ~behavior:"numeric cursor" in + T.require (Server_cursor.compare a b < 0) "numeric comparison reversed"; + T.require (Server_cursor.equal a a) "numeric equality changed") + [ 0L, 1L; Int64.pred Int64.max_int, Int64.max_int ] +;; + let pure_cases = [ Alcotest.test_case + "numeric cursor construction and comparison" + `Quick + numeric_cursor_construction_and_comparison + ; Alcotest.test_case "revision codecs reject unknown versions" `Quick revision_codecs_reject_unknown_versions diff --git a/logseq_overlay_db/test/test_support.ml b/logseq_overlay_db/test/test_support.ml index 1744222..5d6c44b 100644 --- a/logseq_overlay_db/test/test_support.ml +++ b/logseq_overlay_db/test/test_support.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + module Database = Logseq_overlay_db.Database module Graph = Logseq_db_types.Graph_types module Types = Logseq_overlay_db.Types @@ -235,7 +239,7 @@ let seed_mirror ?(graph_id = graph_uuid) support = Logseq_db_types.Sync_checkpoint.create ~graph_id ~schema:Graph.{ major = 65; minor = 33 } - ~applied_server_t:0 + ~applied_server_t:(server_cursor_of_int 0) ~checksum:"0000000000000000" |> require_ok ~behavior:"construct fixture checkpoint" in diff --git a/logseq_overlay_db/tool/fixture_generator.ml b/logseq_overlay_db/tool/fixture_generator.ml index 36ebe05..d0d8c59 100644 --- a/logseq_overlay_db/tool/fixture_generator.ml +++ b/logseq_overlay_db/tool/fixture_generator.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + module Graph = Logseq_db_types.Graph_types module Types = Logseq_overlay_db.Types @@ -162,7 +166,7 @@ let seed_mirror ~application_support_directory ~block_count = Logseq_db_types.Sync_checkpoint.create ~graph_id ~schema:Graph.{ major = 65; minor = 33 } - ~applied_server_t:0 + ~applied_server_t:(server_cursor_of_int 0) ~checksum:"0000000000000000" |> Result.get_ok in diff --git a/logseq_overlay_db/tool/performance_benchmark.ml b/logseq_overlay_db/tool/performance_benchmark.ml index 457f65e..c58b4e9 100644 --- a/logseq_overlay_db/tool/performance_benchmark.ml +++ b/logseq_overlay_db/tool/performance_benchmark.ml @@ -137,19 +137,6 @@ let measure_journals database = then fail "measured get_journals cardinality mismatch")) ;; -let server_cursor value = - Types.Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" value) - |> get_ok "server cursor" -;; - -let server_cursor_number cursor = - Types.Server_cursor.to_string cursor - |> String.split_on_char ':' - |> List.rev - |> List.hd - |> int_of_string -;; - let authoritative_wire operations = let module Transit = Transit_core.Json in let module Codec = Transit_native.Transit.Json in @@ -167,15 +154,13 @@ let block_lookup uuid = ;; let commit_authoritative database ~cursor wire = - let transaction = - Types.authoritative_transaction ~cursor:(server_cursor cursor) ~transaction:wire - in + let transaction = Types.authoritative_transaction ~cursor ~transaction:wire in let batch = Types.authoritative_batch ~maximum_count:16 ~maximum_bytes:(8 * 1_024 * 1_024) ~transactions:[ transaction ] - ~through:(server_cursor cursor) + ~through:cursor ~checksum:None |> get_ok "authoritative batch" in @@ -186,10 +171,16 @@ let commit_authoritative database ~cursor wire = with | Ok value -> value | Error Types.Authoritative_cursor_discontinuous -> - fail "begin authoritative cursor %d is discontinuous" cursor + fail + "begin authoritative cursor %Ld is discontinuous" + (Types.Server_cursor.to_int64 cursor) | Error (Types.Authoritative_decode_failed message) -> - fail "begin authoritative cursor %d decode failed: %s" cursor message - | Error _ -> fail "begin authoritative cursor %d failed" cursor + fail + "begin authoritative cursor %Ld decode failed: %s" + (Types.Server_cursor.to_int64 cursor) + message + | Error _ -> + fail "begin authoritative cursor %Ld failed" (Types.Server_cursor.to_int64 cursor) in if Option.is_some crypto then fail "performance authoritative transaction requested crypto"; @@ -203,14 +194,16 @@ let commit_authoritative database ~cursor wire = let measure_rebase database cursor = let module Transit = Transit_core.Json in - incr cursor; + cursor + := Types.Server_cursor.of_int64 (Int64.succ (Types.Server_cursor.to_int64 !cursor)) + |> Result.get_ok; let wire = authoritative_wire [ Transit.Array [ Transit.Keyword "db/add" ; Transit.Keyword "db/current-tx" ; Transit.Keyword "logseq-overlay/performance-rebase" - ; Transit.Int !cursor + ; Transit.Int64 (Types.Server_cursor.to_int64 !cursor) ] ] in @@ -219,7 +212,9 @@ let measure_rebase database cursor = let measure_single_block_classification database cursor ordinal = let module Transit = Transit_core.Json in - incr cursor; + cursor + := Types.Server_cursor.of_int64 (Int64.succ (Types.Server_cursor.to_int64 !cursor)) + |> Result.get_ok; let block = Fixture_generator.block_uuid ordinal in let wire = authoritative_wire @@ -227,7 +222,8 @@ let measure_single_block_classification database cursor ordinal = [ Transit.Keyword "db/add" ; block_lookup block ; Transit.Keyword "block/updated-at" - ; Transit.Int (1_704_067_300_000 + !cursor) + ; Transit.Int64 + (Int64.add 1_704_067_300_000L (Types.Server_cursor.to_int64 !cursor)) ] ] in @@ -273,14 +269,17 @@ let measure_delete_conflict database cursor ordinal = |> get_ok "commit performance delete" : Types.local_commit_outcome); let module Transit = Transit_core.Json in - incr cursor; + cursor + := Types.Server_cursor.of_int64 (Int64.succ (Types.Server_cursor.to_int64 !cursor)) + |> Result.get_ok; let wire = authoritative_wire [ Transit.Array [ Transit.Keyword "db/add" ; block_lookup block ; Transit.Keyword "block/updated-at" - ; Transit.Int (1_704_067_400_000 + !cursor) + ; Transit.Int64 + (Int64.add 1_704_067_400_000L (Types.Server_cursor.to_int64 !cursor)) ] ] in @@ -323,7 +322,6 @@ let benchmark ~open_database ~database_path ~block_count required_sizes = Database.inspect_sync database |> get_ok "benchmark checkpoint" |> Types.sync_view_checkpoint - |> server_cursor_number |> ref in let single_block_samples = diff --git a/logseq_overlay_db/tool/read_reuse_probe.ml b/logseq_overlay_db/tool/read_reuse_probe.ml index dc6cfcb..ed8ee4d 100644 --- a/logseq_overlay_db/tool/read_reuse_probe.ml +++ b/logseq_overlay_db/tool/read_reuse_probe.ml @@ -116,7 +116,9 @@ let commit db name expected mutation = let token db = D.inspect_sync db |> ok "sync" |> T.sync_view_token let cursor n = - T.Server_cursor.of_string ("server-cursor:v1:" ^ string_of_int n) |> ok "cursor" + T.Server_cursor.of_int64 (Int64.of_int n) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> ok "cursor" ;; let authoritative db n ops = diff --git a/logseq_overlay_db/tool/snapshot_checksum_probe.ml b/logseq_overlay_db/tool/snapshot_checksum_probe.ml index 5fc32fa..3c56879 100644 --- a/logseq_overlay_db/tool/snapshot_checksum_probe.ml +++ b/logseq_overlay_db/tool/snapshot_checksum_probe.ml @@ -170,7 +170,9 @@ let sample ~verify ~root ~name ~graph_id ~path ~expected_rows ~expected ~retry i inspection ~path ~applied_server_cursor: - (T.Server_cursor.of_string "server-cursor:v1:0" |> ok "cursor") + (T.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> ok "cursor") ~expected_checksum:expected ~expected_rows |> ok "prepare") diff --git a/logseq_sync/lib/effect_runner/eio/http.ml b/logseq_sync/lib/effect_runner/eio/http.ml index e0bb2d7..b217b18 100644 --- a/logseq_sync/lib/effect_runner/eio/http.ml +++ b/logseq_sync/lib/effect_runner/eio/http.ml @@ -67,7 +67,7 @@ let snapshot_baseline ~base_url ~graph_id ~token = ~path:(graph_path graph_id "pull") ~query:[] ~token - ~maximum_response_bytes:(64 * 1024 * 1024) + ~maximum_response_bytes:Limits.maximum_snapshot_baseline_response_bytes () ;; diff --git a/logseq_sync/lib/pure_reducer/core.ml b/logseq_sync/lib/pure_reducer/core.ml index 5430129..d5d8a4a 100644 --- a/logseq_sync/lib/pure_reducer/core.ml +++ b/logseq_sync/lib/pure_reducer/core.ml @@ -1,4 +1,5 @@ module Overlay = Logseq_overlay_db.Types +module Server_cursor = Overlay.Server_cursor type graph_id = Logseq_db_types.Graph_types.Uuid.t type graph = Logseq_db_types.Managed_graph.t @@ -50,7 +51,7 @@ type snapshot = { sync_phase : sync_phase ; catalog : graph list ; selected_graph : graph_id option - ; applied_server_t : int option + ; applied_server_t : Server_cursor.t option ; timeline_presentation_pending : bool ; startup : startup_facts ; last_error : string option @@ -560,7 +561,7 @@ type mirror_request = type snapshot_activation_request = { artifact : staged_artifact ; scope : graph_scope - ; applied_server_t : int + ; applied_server_t : Server_cursor.t ; key : graph_key_handle option } @@ -699,7 +700,7 @@ type submission_owner = { batch_id : Overlay.submission_batch_id ; mutation_ids : graph_id list ; connection : connection_scope - ; accepted_through : int option + ; accepted_through : Server_cursor.t option ; response_timer : timer_id option } @@ -710,7 +711,7 @@ type pull_stage = type pull_owner = { pull_connection : connection_scope - ; issued_since : int + ; issued_since : Server_cursor.t ; pull_stage : pull_stage } @@ -744,14 +745,14 @@ type t = ; pending_encrypted_graph_key : string option ; pending_private_key_package : string option ; snapshot_scope : graph_scope option - ; snapshot_server_t : int option + ; snapshot_server_t : Server_cursor.t option ; active_snapshot_download : graph_scope option ; sync_view : Overlay.sync_view option ; pending_sync_inspection : graph_scope option ; pending_outbox_transition : outbox_transition_request option ; submission_owner : submission_owner option ; pull_owner : pull_owner option - ; pull_wanted : (graph_scope * int) option + ; pull_wanted : (graph_scope * Server_cursor.t) option ; pull_retry_delay : float ; active_authoritative_batch : authoritative_batch option ; queued_authoritative_batch : authoritative_batch option @@ -989,30 +990,11 @@ let start_websocket core graph = { next; effects = [ publish next; Run (Start_websocket { scope; uri }) ] } ;; -let server_cursor value = - Overlay.Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" value) -;; - let checksum value = Overlay.Checksum.of_string ("checksum:v1:" ^ value) -let server_cursor_number cursor = - Overlay.Server_cursor.to_string cursor - |> String.split_on_char ':' - |> List.rev - |> List.hd - |> int_of_string -;; - let decode_snapshot_baseline source = - try - match Yojson.Safe.from_string source with - | `Assoc fields -> - (match List.assoc_opt "type" fields, List.assoc_opt "t" fields with - | Some (`String "pull/ok"), Some (`Int server_t) when server_t >= 0 -> Ok server_t - | _ -> Error "snapshot baseline pull is invalid") - | _ -> Error "snapshot baseline pull must be an object" - with - | Yojson.Json_error _ -> Error "snapshot baseline pull is not valid JSON" + Sync_protocol.decode_snapshot_baseline source + |> Result.map_error Sync_protocol.error_to_string ;; let decode_snapshot_uri source = @@ -1379,8 +1361,7 @@ let graph_attached core (attachment : graph_attachment) = let snapshot = { core.public_state.snapshot with sync_phase = Current - ; applied_server_t = - Some (server_cursor_number (Overlay.sync_view_checkpoint attachment.sync)) + ; applied_server_t = Some (Overlay.sync_view_checkpoint attachment.sync) ; timeline_presentation_pending = true ; startup ; last_error = None @@ -1414,14 +1395,14 @@ let submission_message batch = Sync_protocol.Client.Tx_batch { client_revision = Some (Overlay.Submission_batch_id.to_string (Overlay.submission_batch_id batch)) - ; t_before = server_cursor_number (Overlay.submission_batch_t_before batch) + ; t_before = Overlay.submission_batch_t_before batch ; txs } ;; let pull_target_pending core graph cursor = match core.pull_wanted with - | Some (scope, target) -> scope = graph && target > cursor + | Some (scope, target) -> scope = graph && Server_cursor.compare target cursor > 0 | None -> false ;; @@ -1583,7 +1564,8 @@ let note_pull_target core graph through = in let target = match previous, through with - | Some old, Some target -> Some (max old target) + | Some old, Some target -> + Some (if Server_cursor.compare old target >= 0 then old else target) | None, target | target, None -> target in { core with pull_wanted = Option.map (fun target -> graph, target) target } @@ -1642,7 +1624,9 @@ let request_pull ?through ?(force = false) core = ;; let pull_no_progress core (connection : connection_scope) = - let cursor = Option.value core.public_state.snapshot.applied_server_t ~default:0 in + let cursor = + Option.value core.public_state.snapshot.applied_server_t ~default:Server_cursor.zero + in let core = { core with pull_owner = None } in if not (pull_target_pending core connection.graph cursor) then unchanged { core with pull_retry_delay = 1. } @@ -1762,14 +1746,11 @@ let outbox_transition_applied core (result : outbox_transition_result) = | Overlay.Accept_group _ | Reject_group _ -> let through = match result.commit.transition with - | Overlay.Accept_group { barrier; _ } -> - Some (server_cursor_number barrier.through) - | Reject_group { resolution = Stale { through }; _ } -> - Some (server_cursor_number through) + | Overlay.Accept_group { barrier; _ } -> Some barrier.through + | Reject_group { resolution = Stale { through }; _ } -> Some through | Reject_group { resolution = Definitive { partition; _ }; _ } -> Option.map - (fun (barrier : Overlay.acceptance_barrier) -> - server_cursor_number barrier.through) + (fun (barrier : Overlay.acceptance_barrier) -> barrier.through) partition.acceptance_barrier | Submit_group _ | Retry_group _ -> None in @@ -1807,42 +1788,51 @@ let websocket_opened core connection = ;; let authoritative_input core t checksum_value txs = - let current = Option.value core.public_state.snapshot.applied_server_t ~default:(-1) in - let txs = List.filter (fun tx -> tx.Sync_protocol.Server.t > current) txs in - if t >= current && txs = [] + let current = core.public_state.snapshot.applied_server_t in + let ahead cursor = + Option.fold + ~none:true + ~some:(fun current -> Server_cursor.compare cursor current > 0) + current + in + let txs = List.filter (fun tx -> ahead tx.Sync_protocol.Server.t) txs in + if + txs = [] + && Option.fold + ~none:true + ~some:(fun current -> Server_cursor.compare t current >= 0) + current then Ok None else ( let rec encode acc = function | [] -> Ok (List.rev acc) | tx :: rest -> - (match server_cursor tx.Sync_protocol.Server.t with - | Error message -> Error message - | Ok cursor -> - (match - Overlay.encoded_transaction_of_string - ~maximum_bytes:core.config.limits.maximum_response_bytes - tx.tx - with - | Error _ -> Error "invalid encoded transaction" - | Ok transaction -> - encode (Overlay.authoritative_transaction ~cursor ~transaction :: acc) rest)) + (match + Overlay.encoded_transaction_of_string + ~maximum_bytes:core.config.limits.maximum_response_bytes + tx.Sync_protocol.Server.tx + with + | Error _ -> Error "invalid encoded transaction" + | Ok transaction -> + encode + (Overlay.authoritative_transaction ~cursor:tx.t ~transaction :: acc) + rest) in Result.bind (encode [] txs) (fun transactions -> - Result.bind (server_cursor t) (fun through -> - let checksum = - match checksum_value with - | None -> Ok None - | Some value -> Result.map Option.some (checksum value) - in - Result.bind checksum (fun checksum -> - Overlay.authoritative_batch - ~maximum_count:4096 - ~maximum_bytes:core.config.limits.maximum_response_bytes - ~transactions - ~through - ~checksum - |> Result.map Option.some - |> Result.map_error (fun _ -> "invalid authoritative batch"))))) + let checksum = + match checksum_value with + | None -> Ok None + | Some value -> Result.map Option.some (checksum value) + in + Result.bind checksum (fun checksum -> + Overlay.authoritative_batch + ~maximum_count:4096 + ~maximum_bytes:core.config.limits.maximum_response_bytes + ~transactions + ~through:t + ~checksum + |> Result.map Option.some + |> Result.map_error (fun _ -> "invalid authoritative batch")))) ;; let enqueue_authoritative core batch = @@ -1856,8 +1846,9 @@ let enqueue_authoritative core batch = ;; let apply_accept core owner t checksum_value = - match core.sync_view, checksum_value, server_cursor t with - | Some sync, Some checksum_value, Ok through -> + match core.sync_view, checksum_value with + | Some sync, Some checksum_value -> + let through = t in (match checksum checksum_value with | Error _ -> fail core During_catalog "invalid acceptance checksum" | Ok checksum -> @@ -1912,10 +1903,9 @@ let apply_rejection core owner rejection = | _ :: _ -> (match rejection.t, rejection.checksum with | Some through, Some checksum_value -> - Result.bind (server_cursor through) (fun through -> - Result.map - (fun checksum -> Some Overlay.{ through; checksum }) - (checksum checksum_value)) + Result.map + (fun checksum -> Some Overlay.{ through; checksum }) + (checksum checksum_value) | _ -> Error "partial rejection omitted its acceptance barrier") in let reason = @@ -1929,8 +1919,7 @@ let apply_rejection core owner rejection = in let resolution = match rejection.Sync_protocol.reason, rejection.t with - | Sync_protocol.Stale, Some through -> - Result.map (fun through -> Overlay.Stale { through }) (server_cursor through) + | Sync_protocol.Stale, Some through -> Ok (Overlay.Stale { through }) | Stale, None -> Error "stale rejection omitted cursor" | _, _ -> Result.bind (partition ()) (fun (failed_member, unexecuted_suffix) -> @@ -1992,7 +1981,7 @@ let websocket_message core connection message = ~none:(Option.is_none core.active_authoritative_batch) ~some:(fun owner -> owner.pull_connection = connection - && t >= owner.issued_since + && Server_cursor.compare t owner.issued_since >= 0 && match owner.pull_stage with | Awaiting_pull_response _ -> true @@ -2057,7 +2046,7 @@ let authoritative_applied core (result : authoritative_commit_result) = || Option.is_none core.active_authoritative_batch then unchanged core else ( - let applied_server_t = server_cursor_number result.commit.checkpoint in + let applied_server_t = result.commit.checkpoint in let receipt_mutation_id (receipt : Overlay.mutation_receipt) = match receipt with | Applied_receipt { mutation_id; _ } @@ -2093,7 +2082,7 @@ let authoritative_applied core (result : authoritative_commit_result) = then None else ( match owner.accepted_through with - | Some through when applied_server_t >= through -> None + | Some through when Server_cursor.compare applied_server_t through >= 0 -> None | _ -> Some owner) | None -> None in diff --git a/logseq_sync/lib/pure_reducer/sync_protocol.ml b/logseq_sync/lib/pure_reducer/sync_protocol.ml index 249bbef..a186091 100644 --- a/logseq_sync/lib/pure_reducer/sync_protocol.ml +++ b/logseq_sync/lib/pure_reducer/sync_protocol.ml @@ -1,4 +1,4 @@ -type cursor = int +type cursor = Logseq_db_types.Server_cursor.t type checksum = string type rejection_reason = @@ -231,22 +231,47 @@ let required_nullable_string context path ?maximum name fields = Ok (Some value) ;; -let non_negative_int context path = function - | `Int value when value >= 0 -> Ok value - | `Int _ -> error context path (Invalid_field { expected = "non-negative integer" }) - | _ -> error context path (Invalid_field { expected = "non-negative integer" }) +let maximum_wire_cursor = 9007199254740991L + +let cursor_of_json context path value = + let number = + match value with + | `Int value -> Some (Int64.of_int value) + | `Intlit value + when String.length value > 0 + && String.for_all + (function + | '0' .. '9' -> true + | _ -> false) + value -> Int64.of_string_opt value + | _ -> None + in + match number with + | Some value when value >= 0L && value <= maximum_wire_cursor -> + (match Logseq_db_types.Server_cursor.of_int64 value with + | Ok cursor -> Ok cursor + | Error `Negative_cursor -> assert false) + | _ -> + error + context + path + (Invalid_field { expected = "integer cursor between 0 and 9007199254740991" }) +;; + +let cursor_json value = + `Intlit (Int64.to_string (Logseq_db_types.Server_cursor.to_int64 value)) ;; let required_cursor context path name fields = let* value = required_field context path name fields in - non_negative_int context (path @ [ name ]) value + cursor_of_json context (path @ [ name ]) value ;; let optional_cursor context path name fields = match field name fields with | None -> Ok None | Some value -> - let* value = non_negative_int context (path @ [ name ]) value in + let* value = cursor_of_json context (path @ [ name ]) value in Ok (Some value) ;; @@ -503,9 +528,9 @@ let decode_server_fields context message_type fields = | unsupported -> error context [] (Unsupported_message_type unsupported) ;; -let decode direction decode_fields wire = +let decode ?maximum direction decode_fields wire = let context = { direction; message_type = None } in - let maximum = maximum_wire_bytes direction in + let maximum = Option.value maximum ~default:(maximum_wire_bytes direction) in if String.length wire > maximum then error context [] (Limit_exceeded { maximum }) else ( @@ -520,6 +545,16 @@ let decode direction decode_fields wire = | Yojson.Json_error _ -> error context [] Invalid_json) ;; +let decode_snapshot_baseline = + decode + ~maximum:Logseq_db_types.Limits.maximum_snapshot_baseline_response_bytes + Server + (fun context message_type fields -> + if message_type = "pull/ok" + then required_cursor context [] "t" fields + else error context [] (Unsupported_message_type message_type)) +;; + let decode_client_message = decode Client decode_client_fields let decode_server_message = decode Server decode_server_fields let uuid_json uuid = `String (Logseq_db_types.Graph_types.Uuid.to_string uuid) @@ -557,7 +592,7 @@ let client_message_json = function `Assoc fields | Client.Pull { since } -> let fields = [ "type", `String "pull" ] in - `Assoc (add_optional "since" (fun value -> `Int value) since fields) + `Assoc (add_optional "since" cursor_json since fields) | Client.Tx_batch { client_revision; t_before; txs } -> let fields = [ "type", `String "tx/batch" ] in let fields = @@ -565,7 +600,7 @@ let client_message_json = function in `Assoc (fields - @ [ "t-before", `Int t_before + @ [ "t-before", cursor_json t_before ; "txs", `List (List.map client_transaction_json txs) ]) | Client.Ping -> `Assoc [ "type", `String "ping" ] @@ -580,7 +615,7 @@ let user_presence_json (user : user_presence) = ;; let pull_transaction_json (transaction : Server.pull_transaction) = - [ "t", `Int transaction.Server.t; "tx", `String transaction.tx ] + [ "t", cursor_json transaction.Server.t; "tx", `String transaction.tx ] |> add_optional "outliner-op" (fun value -> `String value) transaction.outliner_op |> fun fields -> `Assoc fields ;; @@ -604,7 +639,7 @@ let rejection_json (rejection : rejection) = [ "type", `String "tx/reject" ; "reason", `String (rejection_reason_json rejection.reason) ] - |> add_optional "t" (fun value -> `Int value) rejection.t + |> add_optional "t" cursor_json rejection.t |> add_optional "checksum" (fun value -> `String value) rejection.checksum |> add_non_empty_list "success-tx-ids" uuid_json rejection.success_tx_ids |> add_optional "failed-tx-id" uuid_json rejection.failed_tx_id @@ -616,7 +651,7 @@ let rejection_json (rejection : rejection) = let server_message_json = function | Server.Hello { t; checksum } -> - let fields = [ "type", `String "hello"; "t", `Int t ] in + let fields = [ "type", `String "hello"; "t", cursor_json t ] in `Assoc (add_optional "checksum" (fun value -> `String value) checksum fields) | Server.Online_users { online_users } -> `Assoc @@ -630,13 +665,13 @@ let server_message_json = function ; "editing-block-uuid", nullable_string_json editing_block_uuid ] | Server.Pull_ok { t; checksum; txs } -> - let fields = [ "type", `String "pull/ok"; "t", `Int t ] in + let fields = [ "type", `String "pull/ok"; "t", cursor_json t ] in let fields = add_optional "checksum" (fun value -> `String value) checksum fields in `Assoc (fields @ [ "txs", `List (List.map pull_transaction_json txs) ]) | Server.Tx_batch_ok { t; checksum } -> - let fields = [ "type", `String "tx/batch/ok"; "t", `Int t ] in + let fields = [ "type", `String "tx/batch/ok"; "t", cursor_json t ] in `Assoc (add_optional "checksum" (fun value -> `String value) checksum fields) - | Server.Changed { t } -> `Assoc [ "type", `String "changed"; "t", `Int t ] + | Server.Changed { t } -> `Assoc [ "type", `String "changed"; "t", cursor_json t ] | Server.Tx_reject rejection -> rejection_json rejection | Server.Pong -> `Assoc [ "type", `String "pong" ] | Server.Error { message } -> diff --git a/logseq_sync/spec/pure_reducer/core.mli b/logseq_sync/spec/pure_reducer/core.mli index a488764..f2592a4 100644 --- a/logseq_sync/spec/pure_reducer/core.mli +++ b/logseq_sync/spec/pure_reducer/core.mli @@ -52,7 +52,7 @@ type snapshot = { sync_phase : sync_phase ; catalog : graph list ; selected_graph : graph_id option - ; applied_server_t : int option + ; applied_server_t : Logseq_db_types.Server_cursor.t option ; timeline_presentation_pending : bool ; startup : startup_facts ; last_error : string option @@ -381,7 +381,7 @@ type mirror_request = type snapshot_activation_request = { artifact : staged_artifact ; scope : graph_scope - ; applied_server_t : int + ; applied_server_t : Logseq_db_types.Server_cursor.t ; key : graph_key_handle option } diff --git a/logseq_sync/spec/pure_reducer/sync_protocol.mli b/logseq_sync/spec/pure_reducer/sync_protocol.mli index a6f486a..9c6e8a9 100644 --- a/logseq_sync/spec/pure_reducer/sync_protocol.mli +++ b/logseq_sync/spec/pure_reducer/sync_protocol.mli @@ -1,7 +1,7 @@ (** Typed application-level WebSocket protocol shared by the sync client and server. *) (** A non-negative authoritative transaction cursor. Codec functions reject negative values. *) -type cursor = int +type cursor = Logseq_db_types.Server_cursor.t (** A 16-character hexadecimal entity checksum. *) type checksum = string @@ -137,5 +137,8 @@ val encode_server_message : Server.message -> (string, codec_error) result (** Strictly decode one bounded server JSON message. *) val decode_server_message : string -> (Server.message, codec_error) result +(** Decode the cursor of the HTTP snapshot baseline using the wire range rules. *) +val decode_snapshot_baseline : string -> (cursor, codec_error) result + (** Render safe diagnostic text without including raw payloads or invalid field values. *) val error_to_string : codec_error -> string diff --git a/logseq_sync/test/core_contract.ml b/logseq_sync/test/core_contract.ml index b9f6fd0..cd6fc50 100644 --- a/logseq_sync/test/core_contract.ml +++ b/logseq_sync/test/core_contract.ml @@ -1,3 +1,14 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + +let cursor_testable = + Alcotest.testable + (fun fmt cursor -> + Format.fprintf fmt "%Ld" (Logseq_db_types.Server_cursor.to_int64 cursor)) + Logseq_db_types.Server_cursor.equal +;; + module Core = Logseq_sync_pure_reducer.Core type observed = @@ -440,7 +451,8 @@ let warm_encrypted_graph_loads_key_before_attach_and_protects_outbox () = Logseq_overlay_db.Types.sync_token_of_string "sync-token:v1:warm" |> Result.get_ok in let checkpoint = - Logseq_overlay_db.Types.Server_cursor.of_string "server-cursor:v1:0" + Logseq_overlay_db.Types.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> Result.get_ok in let sync = @@ -1035,7 +1047,10 @@ let cold_start_absent_mirror_reaches_pull_after_snapshot_activation () = activated.effects |> Option.get in - Alcotest.(check int) "activation uses baseline cursor" 42 activation.applied_server_t; + Alcotest.(check cursor_testable) + "activation uses baseline cursor" + (server_cursor_of_int 42) + activation.applied_server_t; Alcotest.(check bool) "plain snapshot has no graph key" true @@ -1060,7 +1075,9 @@ let cold_start_absent_mirror_reaches_pull_after_snapshot_activation () = |> Result.get_ok in let checkpoint = - Logseq_overlay_db.Types.Server_cursor.of_string "server-cursor:v1:42" |> Result.get_ok + Logseq_overlay_db.Types.Server_cursor.of_int64 42L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> Result.get_ok in let sync = Logseq_overlay_db.Types.sync_view ~token:sync_token ~checkpoint ~submissions:[] @@ -1090,6 +1107,66 @@ let cold_start_absent_mirror_reaches_pull_after_snapshot_activation () = pulling.effects) ;; +let snapshot_baseline_above_websocket_budget_bootstraps () = + let requested, _scope = begin_snapshot_bootstrap () in + (* Ordinary cursor values and small valid Transit transactions form a valid + HTTP pull response larger than a WebSocket frame, within the endpoint budget. *) + let transaction = + let module Transit = Transit_core.Json in + Transit_native.Transit.Json.to_string + (Transit.Array + [ Transit.Array + [ Transit.Keyword "db/add" + ; Transit.Int 1 + ; Transit.Keyword "block/title" + ; Transit.String (String.make 16_384 'x') + ] + ]) + in + let body = + Yojson.Safe.to_string + (`Assoc + [ "type", `String "pull/ok" + ; "t", `Int 32 + ; ( "txs" + , `List + (List.init 32 (fun index -> + `Assoc [ "t", `Int (index + 1); "tx", `String transaction ])) ) + ]) + in + Alcotest.(check bool) + (Printf.sprintf + "ordinary %d-byte HTTP fixture exceeds only the WebSocket budget" + (String.length body)) + true + (String.length body > Logseq_db_types.Limits.maximum_response_bytes + && String.length body <= 64 * 1024 * 1024); + let metadata = + List.find_map + (function + | Core.Run (Core.Request (ticket, Core.Fetch_snapshot_baseline _)) -> + Some + (Core.step + requested.next + (Core.Runner_completed (Core.Completion (ticket, Ok body)))) + | Run _ | Delegate _ | Publish _ -> None) + requested.effects + |> Option.get + in + Alcotest.(check bool) + "valid HTTP baseline proceeds to snapshot metadata" + true + (List.exists + (function + | Core.Run (Core.Request (_, Core.Fetch_snapshot_metadata _)) -> true + | Run _ | Delegate _ | Publish _ -> false) + metadata.effects); + Alcotest.(check bool) + "valid HTTP baseline does not fail bootstrap" + false + ((Core.state metadata.next).snapshot.sync_phase = Core.Failed) +;; + let invalid_snapshot_baseline_fails_bootstrap () = let requested, _scope = begin_snapshot_bootstrap () in let failed = @@ -1166,7 +1243,9 @@ let submitted_core () = let scope = attach_request.scope in let sync_token = Overlay.sync_token_of_string "sync-token:v1:1" |> Result.get_ok in let checkpoint = - Overlay.Server_cursor.of_string "server-cursor:v1:0" |> Result.get_ok + Overlay.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> Result.get_ok in let sync = Overlay.sync_view ~token:sync_token ~checkpoint ~submissions:[] in let attached = Core.step inspected.next (Core.Graph_attached { scope; sync }) in @@ -1289,7 +1368,7 @@ let partial_rejection_is_normalized_exactly () = in let rejection : Logseq_sync_pure_reducer.Sync_protocol.rejection = { reason = Db_transact_failed - ; t = Some 1 + ; t = Some (server_cursor_of_int 1) ; checksum = Some "0123456789abcdef" ; success_tx_ids = [ first ] ; failed_tx_id = Some second @@ -1342,7 +1421,9 @@ let accepted_submission_pulls_its_authoritative_transaction () = (Core.step core (Core.Websocket_message - (connection, Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] }))) + ( connection + , Protocol.Server.Pull_ok + { t = server_cursor_of_int 0; checksum = None; txs = [] } ))) .next in let acknowledged = @@ -1351,7 +1432,7 @@ let accepted_submission_pulls_its_authoritative_transaction () = (Core.Websocket_message ( connection , Logseq_sync_pure_reducer.Sync_protocol.Server.Tx_batch_ok - { t = 1; checksum = Some "0123456789abcdef" } )) + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } )) in let request = List.find_map @@ -1365,7 +1446,9 @@ let accepted_submission_pulls_its_authoritative_transaction () = in let sync_token = Overlay.sync_token_of_string "sync-token:v1:2" |> Result.get_ok in let checkpoint = - Overlay.Server_cursor.of_string "server-cursor:v1:0" |> Result.get_ok + Overlay.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> Result.get_ok in let sync = Overlay.sync_view ~token:sync_token ~checkpoint ~submissions:[] in let generation = Overlay.Generation.of_string "generation:v1:2" |> Result.get_ok in @@ -1397,7 +1480,7 @@ let accepted_submission_pulls_its_authoritative_transaction () = (Core.Send_websocket { message = Logseq_sync_pure_reducer.Sync_protocol.Client.Pull { since } ; _ - }) -> since = Some 0 + }) -> since = Some (server_cursor_of_int 0) | Run _ | Delegate _ | Publish _ -> false) completed.effects) ;; @@ -1579,7 +1662,7 @@ let canonical_overlay_happy_path () = selected_observed [ Core.Delegate (Core.Attach_graph mirror_request) ] in - let cursor0 = token Overlay.Server_cursor.of_string "server-cursor:v1:0" in + let cursor0 = token Overlay.Server_cursor.of_int64 0L in let sync_token0 = token Overlay.sync_token_of_string "sync-token:v1:hp-0" in let empty_sync0 = Overlay.sync_view ~token:sync_token0 ~checkpoint:cursor0 ~submissions:[] @@ -1592,7 +1675,7 @@ let canonical_overlay_happy_path () = snapshot = { selected_state.snapshot with sync_phase = Core.Current - ; applied_server_t = Some 0 + ; applied_server_t = Some (server_cursor_of_int 0) ; timeline_presentation_pending = true ; startup = attached_startup } @@ -1645,17 +1728,21 @@ let canonical_overlay_happy_path () = && timer.scope.connection_generation = Some connection.connection_generation && state = pulling0_state && scope = connection - && message = Protocol.Client.Pull { since = Some 0 } -> () + && message = Protocol.Client.Pull { since = Some (server_cursor_of_int 0) } -> + () | _ -> Alcotest.fail "HP08 unexpected instruction shape"); let hp08 = check_step "HP08" hp07.next hp08_event pulling0_observed hp08_preview.effects in let remote_tx : Protocol.Server.pull_transaction = - { t = 1; tx = "[]"; outliner_op = Some "save-block" } + { t = server_cursor_of_int 1; tx = "[]"; outliner_op = Some "save-block" } in let opening_message = Protocol.Server.Pull_ok - { t = 1; checksum = Some "0123456789abcdef"; txs = [ remote_tx ] } + { t = server_cursor_of_int 1 + ; checksum = Some "0123456789abcdef" + ; txs = [ remote_tx ] + } in let hp09_event = Core.Websocket_message (connection, opening_message) in let hp09_preview = preview_step "HP09" hp08.next hp09_event in @@ -1666,9 +1753,9 @@ let canonical_overlay_happy_path () = && request.presentation_generation = 1 && request.lifecycle_generation = 0L && request.key = None - && Overlay.Server_cursor.to_string + && Overlay.Server_cursor.to_int64 (Overlay.authoritative_batch_through request.input) - = "server-cursor:v1:1" + = 1L && List.length (Overlay.authoritative_batch_transactions request.input) = 1 -> request | _ -> Alcotest.fail "HP09 unexpected instruction shape" @@ -1683,7 +1770,7 @@ let canonical_overlay_happy_path () = in let generation1 = token Overlay.Generation.of_string "generation:v1:hp-1" in let projection1 = token Overlay.Projection_revision.of_string "projection:v1:1" in - let cursor1 = token Overlay.Server_cursor.of_string "server-cursor:v1:1" in + let cursor1 = token Overlay.Server_cursor.of_int64 1L in let sync_token1 = token Overlay.sync_token_of_string "sync-token:v1:hp-1" in let sync1 = Overlay.sync_view ~token:sync_token1 ~checkpoint:cursor1 ~submissions:[] in let authoritative_commit1 : Overlay.authoritative_commit = @@ -1703,7 +1790,7 @@ let canonical_overlay_happy_path () = snapshot = { pulling0_state.snapshot with sync_phase = Core.Current - ; applied_server_t = Some 1 + ; applied_server_t = Some (server_cursor_of_int 1) } } in @@ -1820,7 +1907,7 @@ let canonical_overlay_happy_path () = let tx_message = Protocol.Client.Tx_batch { client_revision = Some (Overlay.Submission_batch_id.to_string batch_id) - ; t_before = 1 + ; t_before = server_cursor_of_int 1 ; txs = [ { Protocol.Client.tx = "protected-hp-transaction" ; tx_id = Some mutation_id @@ -1850,7 +1937,8 @@ let canonical_overlay_happy_path () = let hp14_event = Core.Websocket_message ( connection - , Protocol.Server.Tx_batch_ok { t = 2; checksum = Some "fedcba9876543210" } ) + , Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 2; checksum = Some "fedcba9876543210" } ) in let hp14_preview = preview_step "HP14" hp13.next hp14_event in let accept_request = @@ -1863,8 +1951,7 @@ let canonical_overlay_happy_path () = = Overlay.Accept_group { batch_id ; barrier = - { through = - token Overlay.Server_cursor.of_string "server-cursor:v1:2" + { through = token Overlay.Server_cursor.of_int64 2L ; checksum = token Overlay.Checksum.of_string "checksum:v1:fedcba9876543210" } @@ -1922,17 +2009,21 @@ let canonical_overlay_happy_path () = && timer.scope.connection_generation = Some connection.connection_generation && state = pulling1_state && scope = connection - && message = Protocol.Client.Pull { since = Some 1 } -> () + && message = Protocol.Client.Pull { since = Some (server_cursor_of_int 1) } -> + () | _ -> Alcotest.fail "HP15 unexpected instruction shape"); let hp15 = check_step "HP15" hp14.next hp15_event pulling1_observed hp15_preview.effects in let incorporated_tx : Protocol.Server.pull_transaction = - { t = 2; tx = "[]"; outliner_op = Some "save-block" } + { t = server_cursor_of_int 2; tx = "[]"; outliner_op = Some "save-block" } in let confirmation_message = Protocol.Server.Pull_ok - { t = 2; checksum = Some "fedcba9876543210"; txs = [ incorporated_tx ] } + { t = server_cursor_of_int 2 + ; checksum = Some "fedcba9876543210" + ; txs = [ incorporated_tx ] + } in let hp16_event = Core.Websocket_message (connection, confirmation_message) in let hp16_preview = preview_step "HP16" hp15.next hp16_event in @@ -1943,9 +2034,9 @@ let canonical_overlay_happy_path () = && request.presentation_generation = 1 && request.lifecycle_generation = 0L && request.key = None - && Overlay.Server_cursor.to_string + && Overlay.Server_cursor.to_int64 (Overlay.authoritative_batch_through request.input) - = "server-cursor:v1:2" + = 2L && List.length (Overlay.authoritative_batch_transactions request.input) = 1 -> request | _ -> Alcotest.fail "HP16 unexpected instruction shape" @@ -1958,7 +2049,7 @@ let canonical_overlay_happy_path () = pulling1_observed [ Core.Delegate (Core.Apply_authoritative_batch confirmation_batch) ] in - let cursor2 = token Overlay.Server_cursor.of_string "server-cursor:v1:2" in + let cursor2 = token Overlay.Server_cursor.of_int64 2L in let sync_token4 = token Overlay.sync_token_of_string "sync-token:v1:hp-4" in let sync2 = Overlay.sync_view ~token:sync_token4 ~checkpoint:cursor2 ~submissions:[] in let authoritative_commit2 : Overlay.authoritative_commit = @@ -1978,7 +2069,7 @@ let canonical_overlay_happy_path () = snapshot = { pulling1_state.snapshot with sync_phase = Core.Current - ; applied_server_t = Some 2 + ; applied_server_t = Some (server_cursor_of_int 2) } } in @@ -2066,7 +2157,9 @@ let websocket_connect_uses_deployed_graph_path () = let scope = attach_request.scope in let sync_token = Overlay.sync_token_of_string "sync-token:v1:1" |> Result.get_ok in let checkpoint = - Overlay.Server_cursor.of_string "server-cursor:v1:0" |> Result.get_ok + Overlay.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> Result.get_ok in let sync = Overlay.sync_view ~token:sync_token ~checkpoint ~submissions:[] in let attached = Core.step inspected.next (Core.Graph_attached { scope; sync }) in @@ -2325,7 +2418,9 @@ let deletion_ready_graph encrypted = in let token = Overlay.sync_token_of_string "sync-token:v1:delete" |> Result.get_ok in let checkpoint = - Overlay.Server_cursor.of_string "server-cursor:v1:0" |> Result.get_ok + Overlay.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> Result.get_ok in let sync = Overlay.sync_view ~token ~checkpoint ~submissions:[] in let attached = Core.step attaching.next (Core.Graph_attached { scope; sync }) in @@ -2589,7 +2684,8 @@ let local_deletion_discards_unsynchronized_work () = deleting.next [ Core.Websocket_message ( connection - , Protocol.Server.Tx_batch_ok { t = 1; checksum = Some "0123456789abcdef" } ) + , Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } ) ]; let closed = Core.step deleting.next (Core.Graph_detached (connection.graph, Ok ())) in ignore (deletion_mirror closed.effects) @@ -2684,7 +2780,11 @@ let recovery_delete_transaction = let submitted_delete_recovery_fixture () = let mutation_id = List.hd mutation_ids in - let cursor = Overlay.Server_cursor.of_string "server-cursor:v1:0" |> Result.get_ok in + let cursor = + Overlay.Server_cursor.of_int64 0L + |> Result.map_error (fun `Negative_cursor -> "negative cursor") + |> Result.get_ok + in let token ordinal = Overlay.sync_token_of_string (Printf.sprintf "sync-token:v1:%d" ordinal) |> Result.get_ok @@ -2789,10 +2889,10 @@ let defer_recovered_delete core connection batch_id = (Core.Websocket_message ( connection , Protocol.Server.Pull_ok - { t = 1 + { t = server_cursor_of_int 1 ; checksum = Some "0123456789abcdef" ; txs = - [ { t = 1 + [ { t = server_cursor_of_int 1 ; tx = recovery_delete_transaction ; outliner_op = Some "delete-blocks" } @@ -2835,7 +2935,8 @@ let live_submitted_delete_resumes_after_acknowledgement () = deferred (Core.Websocket_message ( connection - , Protocol.Server.Tx_batch_ok { t = 1; checksum = Some "0123456789abcdef" } )) + , Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } )) in let request = List.find_map @@ -2886,7 +2987,8 @@ let live_submitted_delete_resumes_after_acknowledgement () = ;; let recovery_cursor n = - Overlay.Server_cursor.of_string (Printf.sprintf "server-cursor:v1:%d" n) + Overlay.Server_cursor.of_int64 (Int64.of_int n) + |> Result.map_error (fun `Negative_cursor -> "negative cursor") |> Result.get_ok ;; @@ -2997,7 +3099,10 @@ let recover_retry core connection sync wire_batch effects = (Some (Overlay.Submission_batch_id.to_string (Overlay.submission_batch_id wire_batch))) client_revision; - Alcotest.(check int) "retry retains conditional baseline" 0 t_before; + Alcotest.(check cursor_testable) + "retry retains conditional baseline" + (server_cursor_of_int 0) + t_before; Alcotest.(check string) "retry retains protected bytes" recovery_delete_transaction @@ -3022,7 +3127,7 @@ let finish_recovery ~reject core connection sync wire_batch effects batch = then Protocol.Server.Tx_reject { reason = Protocol.Stale - ; t = Some 1 + ; t = Some (server_cursor_of_int 1) ; checksum = Some "0123456789abcdef" ; success_tx_ids = [] ; failed_tx_id = None @@ -3030,7 +3135,9 @@ let finish_recovery ~reject core connection sync wire_batch effects batch = ; error_detail = None ; data = None } - else Protocol.Server.Tx_batch_ok { t = 1; checksum = Some "0123456789abcdef" } + else + Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } in let terminal = Core.step retried.next (Core.Websocket_message (connection, response)) in let request = recovery_request terminal.effects in @@ -3066,9 +3173,9 @@ let finish_recovery ~reject core connection sync wire_batch effects batch = let current = recovery_apply resumed.next connection (recovery_sync sync ~checkpoint:1 5 []) in - Alcotest.(check (option int)) + Alcotest.(check (option cursor_testable)) "recovery advances the durable checkpoint" - (Some 1) + (Some (server_cursor_of_int 1)) (Core.state current.next).snapshot.applied_server_t; Alcotest.(check bool) "completed recovery becomes Current" @@ -3118,9 +3225,9 @@ let submitted_delete_survives_disconnect_reconnect () = let submitted_delete_survives_fresh_core_restore () = let _, _, persisted_sync, wire_batch = submitted_delete_recovery_fixture () in let fresh, connection = restore_recovery_core persisted_sync in - Alcotest.(check (option int)) + Alcotest.(check (option cursor_testable)) "fresh Core restores the durable checkpoint" - (Some 0) + (Some (server_cursor_of_int 0)) (Core.state fresh).snapshot.applied_server_t; check_submitted_delete_recovery fresh connection persisted_sync wire_batch ;; @@ -3147,7 +3254,9 @@ let recovery_empty_pull_prioritizes_submitted () = Core.step fresh (Core.Websocket_message - (connection, Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] })) + ( connection + , Protocol.Server.Pull_ok + { t = server_cursor_of_int 0; checksum = None; txs = [] } )) in require_no_new_submission pulled.effects; ignore (recover_retry pulled.next connection sync wire_batch pulled.effects) @@ -3178,7 +3287,9 @@ let recovery_barrier_blocks_queued ~rejected () = Core.step fresh (Core.Websocket_message - (connection, Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] })) + ( connection + , Protocol.Server.Pull_ok + { t = server_cursor_of_int 0; checksum = None; txs = [] } )) in require_no_new_submission pulled.effects; Alcotest.(check bool) @@ -3218,7 +3329,8 @@ let recovery_stale_connection_and_interrupted_retry () = opened.next (Core.Websocket_message ( old_connection - , Protocol.Server.Tx_batch_ok { t = 1; checksum = Some "0123456789abcdef" } )) + , Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } )) in check_instructions "old response cannot accept recovered batch" [] late.effects; check_submitted_delete_recovery late.next new_connection sync wire_batch @@ -3261,7 +3373,8 @@ let recovery_outbox_failure_is_correlated () = submitted.next (Core.Websocket_message ( connection - , Protocol.Server.Tx_batch_ok { t = 1; checksum = Some "0123456789abcdef" } )) + , Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } )) in let request = recovery_request acknowledged.effects in let message = "The overlay outbox transition was rejected." in @@ -3306,7 +3419,8 @@ let recovery_outbox_failure_is_correlated () = duplicate.next (Core.Websocket_message ( connection - , Protocol.Server.Tx_batch_ok { t = 1; checksum = Some "0123456789abcdef" } )) + , Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } )) in check_instructions "old response cannot recreate a failed transition" [] old_ack.effects; ignore batch @@ -3383,12 +3497,15 @@ let pull_message core connection message = let pull_response core connection ~after through = let txs = List.init (through - after) (fun offset -> - { Protocol.Server.t = after + offset + 1; tx = "[]"; outliner_op = None }) + { Protocol.Server.t = server_cursor_of_int (after + offset + 1) + ; tx = "[]" + ; outliner_op = None + }) in pull_message core connection - (Protocol.Server.Pull_ok { t = through; checksum = None; txs }) + (Protocol.Server.Pull_ok { t = server_cursor_of_int through; checksum = None; txs }) ;; let pull_open_and_notifications_are_singleflight () = @@ -3398,31 +3515,44 @@ let pull_open_and_notifications_are_singleflight () = pull_message opened.next connection - (Protocol.Server.Hello { t = 100; checksum = None }) + (Protocol.Server.Hello { t = server_cursor_of_int 100; checksum = None }) in check_pull_count "hello joins the opening pull" 0 hello.effects; let state = ref hello.next in for t = 1 to 100 do - let changed = pull_message !state connection (Protocol.Server.Changed { t }) in + let changed = + pull_message + !state + connection + (Protocol.Server.Changed { t = server_cursor_of_int t }) + in check_pull_count "notification burst joins the same pull" 0 changed.effects; state := changed.next done; - Alcotest.(check (option int)) + Alcotest.(check (option cursor_testable)) "notifications cannot advance the durable checkpoint" - (Some 0) + (Some (server_cursor_of_int 0)) (Core.state !state).snapshot.applied_server_t ;; let pull_demand_survives_authoritative_apply () = let opened, connection, sync = pull_fixture () in let notified = - pull_message opened.next connection (Protocol.Server.Changed { t = 1 }) + pull_message + opened.next + connection + (Protocol.Server.Changed { t = server_cursor_of_int 1 }) in let received = pull_response notified.next connection ~after:0 1 in let state = ref received.next in for index = 1 to 100 do let t = index + 1 in - let changed = pull_message !state connection (Protocol.Server.Changed { t }) in + let changed = + pull_message + !state + connection + (Protocol.Server.Changed { t = server_cursor_of_int t }) + in check_pull_count "apply owns the pull until durable completion" 0 changed.effects; state := changed.next done; @@ -3432,7 +3562,7 @@ let pull_demand_survives_authoritative_apply () = Alcotest.(check bool) "latest target schedules one pull from committed cursor" true - (pull_sends committed.effects = [ connection, Some 1 ]); + (pull_sends committed.effects = [ connection, Some (server_cursor_of_int 1) ]); Alcotest.(check bool) "catchup owner is represented by Pulling phase" true @@ -3453,33 +3583,45 @@ let pull_demand_survives_authoritative_apply () = Alcotest.(check bool) "partial catchup cannot drop the maximum requested target" true - (pull_sends committed.effects = [ connection, Some 2 ]); - (* A sparse page is permitted by authoritative_batch: ordering plus last = through, - with no contiguous-cursor requirement. One transaction also avoids the existing - lexical Server_cursor.compare issue across decimal digit boundaries. *) + (pull_sends committed.effects = [ connection, Some (server_cursor_of_int 2) ]); + (* This pure policy test uses a sparse page to acknowledge the newest target. + The overlay separately enforces contiguous durable application. *) let received = pull_message committed.next connection (Protocol.Server.Pull_ok - { t = 101 + { t = server_cursor_of_int 101 ; checksum = None - ; txs = [ { Protocol.Server.t = 101; tx = "[]"; outliner_op = None } ] + ; txs = + [ { Protocol.Server.t = server_cursor_of_int 101 + ; tx = "[]" + ; outliner_op = None + } + ] }) in let committed = recovery_apply received.next connection (recovery_sync sync ~checkpoint:101 3 []) in - Alcotest.(check (option int)) + Alcotest.(check (option cursor_testable)) "latest checkpoint is durably acknowledged" - (Some 101) + (Some (server_cursor_of_int 101)) (Core.state committed.next).snapshot.applied_server_t; check_pull_count "caught up exactly to latest target" 0 committed.effects; let stale = - pull_message committed.next connection (Protocol.Server.Changed { t = 100 }) + pull_message + committed.next + connection + (Protocol.Server.Changed { t = server_cursor_of_int 100 }) in check_pull_count "old notifications need no pull" 0 stale.effects; - let same = pull_message stale.next connection (Protocol.Server.Changed { t = 101 }) in + let same = + pull_message + stale.next + connection + (Protocol.Server.Changed { t = server_cursor_of_int 101 }) + in check_pull_count "same cursor notification needs no pull" 0 same.effects ;; @@ -3494,11 +3636,12 @@ let pull_no_progress_is_delayed_and_bounded () = pull_message !state connection - (Protocol.Server.Pull_ok { t = reported_head; checksum = None; txs = [] }) + (Protocol.Server.Pull_ok + { t = server_cursor_of_int reported_head; checksum = None; txs = [] }) in - Alcotest.(check (option int)) + Alcotest.(check (option cursor_testable)) "empty response cannot advance durable checkpoint" - (Some 0) + (Some (server_cursor_of_int 0)) (Core.state empty.next).snapshot.applied_server_t; check_pull_count "no progress cannot immediately repeat the pull" 0 empty.effects; Alcotest.(check bool) @@ -3523,14 +3666,17 @@ let pull_no_progress_is_delayed_and_bounded () = && timer.delay_seconds >= !previous_delay); previous_delay := timer.delay_seconds; let notified = - pull_message empty.next connection (Protocol.Server.Changed { t = 10 }) + pull_message + empty.next + connection + (Protocol.Server.Changed { t = server_cursor_of_int 10 }) in check_pull_count "notifications cannot bypass retry delay" 0 notified.effects; let retried = Core.step notified.next (Core.Timer_elapsed timer.id) in Alcotest.(check bool) "retry retains demand and original durable cursor" true - (pull_sends retried.effects = [ connection, Some 0 ]); + (pull_sends retried.effects = [ connection, Some (server_cursor_of_int 0) ]); let duplicate = Core.step retried.next (Core.Timer_elapsed timer.id) in check_instructions "retry timer is consumed once" [] duplicate.effects; state := duplicate.next @@ -3561,7 +3707,10 @@ let pull_timeout_reconnects_and_fences_old_work () = let old_response = pull_response old_timeout.next connection ~after:0 1 in check_instructions "old connection response is inert" [] old_response.effects; let changed = - pull_message old_response.next replacement (Protocol.Server.Changed { t = 2 }) + pull_message + old_response.next + replacement + (Protocol.Server.Changed { t = server_cursor_of_int 2 }) in check_pull_count "replacement still owns its single pull" 0 changed.effects ;; @@ -3576,7 +3725,10 @@ let pull_response_timer_does_not_interrupt_apply () = [] late_timer.effects; let changed = - pull_message late_timer.next connection (Protocol.Server.Changed { t = 2 }) + pull_message + late_timer.next + connection + (Protocol.Server.Changed { t = server_cursor_of_int 2 }) in check_pull_count "notification waits for durable apply" 0 changed.effects; let committed = @@ -3585,7 +3737,7 @@ let pull_response_timer_does_not_interrupt_apply () = Alcotest.(check bool) "apply completion releases exactly one catchup" true - (pull_sends committed.effects = [ connection, Some 1 ]) + (pull_sends committed.effects = [ connection, Some (server_cursor_of_int 1) ]) ;; let pull_terminal_outcome_joins_existing_round ~reject () = @@ -3595,7 +3747,7 @@ let pull_terminal_outcome_joins_existing_round ~reject () = then Protocol.Server.Tx_reject { reason = Protocol.Stale - ; t = Some 1 + ; t = Some (server_cursor_of_int 1) ; checksum = Some "0123456789abcdef" ; success_tx_ids = [] ; failed_tx_id = None @@ -3603,7 +3755,9 @@ let pull_terminal_outcome_joins_existing_round ~reject () = ; error_detail = None ; data = None } - else Protocol.Server.Tx_batch_ok { t = 1; checksum = Some "0123456789abcdef" } + else + Protocol.Server.Tx_batch_ok + { t = server_cursor_of_int 1; checksum = Some "0123456789abcdef" } in let terminal = pull_message core connection response in let request = recovery_request terminal.effects in @@ -3625,7 +3779,7 @@ let pull_terminal_outcome_joins_existing_round ~reject () = Alcotest.(check bool) "terminal outcome target survives until catchup" true - (pull_sends catchup.effects = [ connection, Some 0 ]) + (pull_sends catchup.effects = [ connection, Some (server_cursor_of_int 0) ]) ;; let pull_terminal_barrier_already_applied_is_current ~reject () = @@ -3674,7 +3828,7 @@ let pull_terminal_barrier_already_applied_is_current ~reject () = state.snapshot.sync_phase = Core.Current | _ -> false) completed.effects)) - [ 0; 1 ] + [ server_cursor_of_int 0; server_cursor_of_int 1 ] ;; let pull_shutdown_and_graph_retirement_cancel_demand () = @@ -3686,7 +3840,10 @@ let pull_shutdown_and_graph_retirement_cancel_demand () = let late_timer = Core.step retired.next (Core.Timer_elapsed timer.id) in check_instructions "retired graph timer is inert" [] late_timer.effects; let late_changed = - pull_message late_timer.next connection (Protocol.Server.Changed { t = 20 }) + pull_message + late_timer.next + connection + (Protocol.Server.Changed { t = server_cursor_of_int 20 }) in check_instructions "retired graph demand is inert" [] late_changed.effects; let late_response = pull_response late_changed.next connection ~after:0 1 in @@ -3791,7 +3948,7 @@ let rejected_submission_feedback () = let core, connection = submitted_core () in let rejection : Protocol.rejection = { reason = Db_transact_failed - ; t = Some 0 + ; t = Some (server_cursor_of_int 0) ; checksum = None ; success_tx_ids = [] ; failed_tx_id = Some (List.hd mutation_ids) @@ -3848,7 +4005,9 @@ let rejected_submission_feedback () = Core.step committed.next (Core.Websocket_message - (connection, Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] })) + ( connection + , Protocol.Server.Pull_ok + { t = server_cursor_of_int 0; checksum = None; txs = [] } )) in Alcotest.(check bool) "empty catch-up does not hide the rejected write" @@ -3861,7 +4020,8 @@ let rejected_submission_feedback () = restored (Core.Websocket_message ( restored_connection - , Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] } )) + , Protocol.Server.Pull_ok + { t = server_cursor_of_int 0; checksum = None; txs = [] } )) in Alcotest.(check bool) "restored blocked writes retain visible feedback" @@ -3874,7 +4034,7 @@ let stale_rejection_is_not_a_failed_save_notice () = let core, connection = submitted_core () in let rejection : Protocol.rejection = { reason = Stale - ; t = Some 1 + ; t = Some (server_cursor_of_int 1) ; checksum = None ; success_tx_ids = [] ; failed_tx_id = None @@ -3900,8 +4060,89 @@ let stale_rejection_is_not_a_failed_save_notice () = | _ -> false) ;; +let decimal_boundary_pull_is_admitted () = + List.iter + (fun (first, last) -> + let checkpoint = recovery_cursor (first - 1) in + let sync = + Overlay.sync_view + ~token:(token Overlay.sync_token_of_string "sync-token:v1:numeric-cursor") + ~checkpoint + ~submissions:[] + in + let core, connection = restore_recovery_core sync in + let txs = + List.map + (fun t -> + Protocol.Server. + { t = server_cursor_of_int t; tx = "[]"; outliner_op = None }) + [ first; last ] + in + let wire = + Protocol.encode_server_message + (Protocol.Server.Pull_ok + { t = server_cursor_of_int last; checksum = None; txs }) + |> Result.get_ok + in + let message = Protocol.decode_server_message wire |> Result.get_ok in + let received = Core.step core (Core.Websocket_message (connection, message)) in + Alcotest.(check bool) + "normal numeric pull reaches authoritative apply" + true + (List.exists + (function + | Core.Delegate (Core.Apply_authoritative_batch _) -> true + | _ -> false) + received.effects); + Alcotest.(check bool) + "normal numeric pull does not fail" + false + ((Core.state received.next).snapshot.sync_phase = Core.Failed)) + [ 9, 10; 99, 100; 999, 1000 ] +;; + +let unordered_pull_is_rejected () = + List.iter + (fun values -> + let opened, connection, _ = pull_fixture () in + let txs = + List.map + (fun t -> + Protocol.Server. + { t = server_cursor_of_int t; tx = "[]"; outliner_op = None }) + values + in + let received = + pull_message + opened.next + connection + (Protocol.Server.Pull_ok + { t = server_cursor_of_int (List.hd (List.rev values)) + ; checksum = None + ; txs + }) + in + Alcotest.(check bool) + "duplicate or reversed pull fails" + true + ((Core.state received.next).snapshot.sync_phase = Core.Failed)) + [ [ 9; 9 ]; [ 10; 9 ] ] +;; + let scenarios = - [ Alcotest.test_case "rejected submission feedback" `Quick rejected_submission_feedback + [ Alcotest.test_case + "snapshot baseline retains HTTP response budget" + `Quick + snapshot_baseline_above_websocket_budget_bootstraps + ; Alcotest.test_case + "numeric cursor decimal boundaries" + `Quick + decimal_boundary_pull_is_admitted + ; Alcotest.test_case + "numeric cursor duplicate and reverse order" + `Quick + unordered_pull_is_rejected + ; Alcotest.test_case "rejected submission feedback" `Quick rejected_submission_feedback ; Alcotest.test_case "stale rejection remains recoverable" `Quick diff --git a/logseq_sync/test/pure_reducer_bad_case_03.ml b/logseq_sync/test/pure_reducer_bad_case_03.ml index 9d1ea17..23cbc80 100644 --- a/logseq_sync/test/pure_reducer_bad_case_03.ml +++ b/logseq_sync/test/pure_reducer_bad_case_03.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + open Pure_reducer_bad_case_support (* Scenario: Graph attachment is replayed after the WebSocket is already live. @@ -9,7 +13,9 @@ let run () = let rejected = check_step "BC03" fixture.opened.next event (observe fixture.opened.next) [] in - let message = Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] } in + let message = + Protocol.Server.Pull_ok { t = server_cursor_of_int 0; checksum = None; txs = [] } + in let follow_up = Core.Websocket_message (fixture.connection, message) in let accepted = Core.step rejected.next follow_up diff --git a/logseq_sync/test/pure_reducer_bad_case_04.ml b/logseq_sync/test/pure_reducer_bad_case_04.ml index 6a52576..c5b7297 100644 --- a/logseq_sync/test/pure_reducer_bad_case_04.ml +++ b/logseq_sync/test/pure_reducer_bad_case_04.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + open Pure_reducer_bad_case_support (* Scenario: WebSocket_opened is replayed for an already live connection. The @@ -9,7 +13,9 @@ let run () = let rejected = check_step "BC04" fixture.opened.next event (observe fixture.opened.next) [] in - let message = Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] } in + let message = + Protocol.Server.Pull_ok { t = server_cursor_of_int 0; checksum = None; txs = [] } + in let follow_up = Core.Websocket_message (fixture.connection, message) in let accepted = Core.step rejected.next follow_up diff --git a/logseq_sync/test/pure_reducer_bad_case_05.ml b/logseq_sync/test/pure_reducer_bad_case_05.ml index b2a7a72..0e05f13 100644 --- a/logseq_sync/test/pure_reducer_bad_case_05.ml +++ b/logseq_sync/test/pure_reducer_bad_case_05.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + open Pure_reducer_bad_case_support (* Scenario: A server message arrives after its WebSocket connection closed. @@ -9,7 +13,9 @@ let run () = let closed = Core.step fixture.opened.next closed_event in let late_message = Core.Websocket_message - (fixture.connection, Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] }) + ( fixture.connection + , Protocol.Server.Pull_ok { t = server_cursor_of_int 0; checksum = None; txs = [] } + ) in let rejected = check_step "BC05" closed.next late_message (observe closed.next) [] in let recovery_event = diff --git a/logseq_sync/test/pure_reducer_bad_case_06.ml b/logseq_sync/test/pure_reducer_bad_case_06.ml index 34d6097..e24adc5 100644 --- a/logseq_sync/test/pure_reducer_bad_case_06.ml +++ b/logseq_sync/test/pure_reducer_bad_case_06.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + open Pure_reducer_bad_case_support (* Scenario: An authoritative apply completion arrives without a pending apply. @@ -6,7 +10,7 @@ open Pure_reducer_bad_case_support let run () = let fixture = connected_fixture () in let scope = fixture.connection.graph in - let cursor1 = token Overlay.Server_cursor.of_string "server-cursor:v1:1" in + let cursor1 = token Overlay.Server_cursor.of_int64 1L in let sync_token = token Overlay.sync_token_of_string "sync-token:v1:bc06" in let generation = token Overlay.Generation.of_string "generation:v1:6" in let projection = token Overlay.Projection_revision.of_string "projection:v1:6" in @@ -28,12 +32,13 @@ let run () = check_step "BC06" fixture.opened.next completion (observe fixture.opened.next) [] in let remote_tx : Protocol.Server.pull_transaction = - { t = 1; tx = "[]"; outliner_op = None } + { t = server_cursor_of_int 1; tx = "[]"; outliner_op = None } in let request_event = Core.Websocket_message ( fixture.connection - , Protocol.Server.Pull_ok { t = 1; checksum = None; txs = [ remote_tx ] } ) + , Protocol.Server.Pull_ok + { t = server_cursor_of_int 1; checksum = None; txs = [ remote_tx ] } ) in let requested = Core.step rejected.next request_event in (match requested.effects with @@ -42,7 +47,7 @@ let run () = let accepted = Core.step requested.next completion in match accepted.effects with | [ Core.Publish (Core.State_changed state) ] - when state.snapshot.applied_server_t = Some 1 -> () + when state.snapshot.applied_server_t = Some (server_cursor_of_int 1) -> () | _ -> Alcotest.fail "BC06 exact authoritative completion was not accepted" ;; diff --git a/logseq_sync/test/pure_reducer_bad_case_09.ml b/logseq_sync/test/pure_reducer_bad_case_09.ml index ce71c63..e9b7fa6 100644 --- a/logseq_sync/test/pure_reducer_bad_case_09.ml +++ b/logseq_sync/test/pure_reducer_bad_case_09.ml @@ -6,7 +6,7 @@ open Pure_reducer_bad_case_support let run () = let fixture, current = current_connected_fixture () in let scope = fixture.connection.graph in - let checkpoint = token Overlay.Server_cursor.of_string "server-cursor:v1:0" in + let checkpoint = token Overlay.Server_cursor.of_int64 0L in let sync_token = token Overlay.sync_token_of_string "sync-token:v1:bc09" in let sync = queued_sync sync_token checkpoint (List.hd mutation_ids) in let event = Core.Sync_inspected { scope; sync } in diff --git a/logseq_sync/test/pure_reducer_bad_case_support.ml b/logseq_sync/test/pure_reducer_bad_case_support.ml index 07e7f12..1cddaf7 100644 --- a/logseq_sync/test/pure_reducer_bad_case_support.ml +++ b/logseq_sync/test/pure_reducer_bad_case_support.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + module Core = Logseq_sync_pure_reducer.Core module Overlay = Logseq_overlay_db.Types module Protocol = Logseq_sync_pure_reducer.Sync_protocol @@ -126,7 +130,7 @@ let connected_fixture () = request | _ -> Alcotest.fail "connected fixture did not emit exact Attach_graph" in - let checkpoint = token Overlay.Server_cursor.of_string "server-cursor:v1:0" in + let checkpoint = token Overlay.Server_cursor.of_int64 0L in let sync_token = token Overlay.sync_token_of_string "sync-token:v1:fixture-0" in let sync = Overlay.sync_view ~token:sync_token ~checkpoint ~submissions:[] in let attachment : Core.graph_attachment = { scope = attach_request.scope; sync } in @@ -141,10 +145,12 @@ let connected_fixture () = (match opened.effects with | [ Core.Publish (Core.State_changed _) ; Core.Run - (Core.Send_websocket { scope; message = Protocol.Client.Pull { since = Some 0 } }) + (Core.Send_websocket + { scope; message = Protocol.Client.Pull { since = Some since } }) ; Core.Run (Core.Schedule_timer timer) ] when scope = connection + && Logseq_db_types.Server_cursor.equal since Logseq_db_types.Server_cursor.zero && timer.delay_seconds = 30. && timer.scope.connection_generation = Some connection.connection_generation -> () @@ -158,7 +164,9 @@ let current_connected_fixture () = Core.step fixture.opened.next (Core.Websocket_message - (fixture.connection, Protocol.Server.Pull_ok { t = 0; checksum = None; txs = [] })) + ( fixture.connection + , Protocol.Server.Pull_ok + { t = server_cursor_of_int 0; checksum = None; txs = [] } )) in (match current.effects with | [ Core.Publish (Core.State_changed state) ] @@ -201,7 +209,7 @@ let pending_submission_fixture () = () | _ -> Alcotest.fail "pending submission fixture did not inspect sync state"); let mutation_id = List.hd mutation_ids in - let checkpoint = token Overlay.Server_cursor.of_string "server-cursor:v1:0" in + let checkpoint = token Overlay.Server_cursor.of_int64 0L in let expected = token Overlay.sync_token_of_string "sync-token:v1:bad-case-before" in let sync = queued_sync expected checkpoint mutation_id in let planned = Core.step changed.next (Core.Sync_inspected { scope; sync }) in diff --git a/logseq_sync/test/sync_protocol_contract.ml b/logseq_sync/test/sync_protocol_contract.ml index f4d33b8..0cf9410 100644 --- a/logseq_sync/test/sync_protocol_contract.ml +++ b/logseq_sync/test/sync_protocol_contract.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + module Protocol = Logseq_sync_pure_reducer.Sync_protocol module Client = Protocol.Client module Server = Protocol.Server @@ -48,11 +52,12 @@ let test_all_client_messages_round_trip () = ; Client.Presence { editing_block_uuid = None } ; Client.Presence { editing_block_uuid = Some "block-1" } ; Client.Pull { since = None } - ; Client.Pull { since = Some 7 } - ; Client.Tx_batch { client_revision = None; t_before = 7; txs = [] } + ; Client.Pull { since = Some (server_cursor_of_int 7) } + ; Client.Tx_batch + { client_revision = None; t_before = server_cursor_of_int 7; txs = [] } ; Client.Tx_batch { client_revision = Some "revision-1" - ; t_before = 7 + ; t_before = server_cursor_of_int 7 ; txs = [ { tx = "transit"; tx_id = Some tx_id; outliner_op = Some "save-block" } ; { tx = "second"; tx_id = None; outliner_op = None } @@ -72,7 +77,8 @@ let test_all_client_messages_round_trip () = Alcotest.(check string) "pull uses canonical since" {|{"type":"pull","since":7}|} - (Protocol.encode_client_message (Client.Pull { since = Some 7 }) + (Protocol.encode_client_message + (Client.Pull { since = Some (server_cursor_of_int 7) }) |> require_ok "encode canonical pull") ;; @@ -100,7 +106,7 @@ let rejection let test_all_server_messages_round_trip () = let messages = - [ Server.Hello { t = 7; checksum = Some "0123456789abcdef" } + [ Server.Hello { t = server_cursor_of_int 7; checksum = Some "0123456789abcdef" } ; Server.Online_users { online_users = [ { user_id = "user-1" @@ -114,12 +120,13 @@ let test_all_server_messages_round_trip () = ; Server.Presence { user_id = "user-1"; editing_block_uuid = None } ; Server.Presence { user_id = "user-1"; editing_block_uuid = Some "block-1" } ; Server.Pull_ok - { t = 8 + { t = server_cursor_of_int 8 ; checksum = None - ; txs = [ { t = 8; tx = "transit"; outliner_op = None } ] + ; txs = [ { t = server_cursor_of_int 8; tx = "transit"; outliner_op = None } ] } - ; Server.Tx_batch_ok { t = 8; checksum = Some "fedcba9876543210" } - ; Server.Changed { t = 9 } + ; Server.Tx_batch_ok + { t = server_cursor_of_int 8; checksum = Some "fedcba9876543210" } + ; Server.Changed { t = server_cursor_of_int 9 } ; Server.Pong ; Server.Error { message = "server error" } ] @@ -136,14 +143,15 @@ let test_all_server_messages_round_trip () = let test_all_rejection_shapes_round_trip () = let messages = - [ Server.Tx_reject (rejection ~t:9 Protocol.Stale) + [ Server.Tx_reject (rejection ~t:(server_cursor_of_int 9) Protocol.Stale) ; Server.Tx_reject (rejection Protocol.Empty_tx_data) ; Server.Tx_reject (rejection Protocol.Invalid_tx) ; Server.Tx_reject (rejection Protocol.Invalid_t_before) - ; Server.Tx_reject (rejection ~t:9 Protocol.Snapshot_upload_in_progress) + ; Server.Tx_reject + (rejection ~t:(server_cursor_of_int 9) Protocol.Snapshot_upload_in_progress) ; Server.Tx_reject (rejection - ~t:10 + ~t:(server_cursor_of_int 10) ~checksum:"0123456789abcdef" ~success_tx_ids:[ tx_id ] ~failed_tx_id @@ -328,7 +336,7 @@ let test_protocol_validation_is_fail_closed () = Alcotest.bool "negative typed cursor cannot be encoded" true - (Result.is_error (Protocol.encode_client_message (Client.Pull { since = Some (-1) }))); + (Result.is_error (Logseq_db_types.Server_cursor.of_int64 (-1L))); let oversized = String.make (Logseq_db_types.Limits.maximum_response_bytes + 1) 'x' in let error = Protocol.decode_server_message oversized |> require_error "oversized server message" @@ -342,8 +350,41 @@ let test_protocol_validation_is_fail_closed () = ) ;; +let test_cursor_wire_range () = + List.iter + (fun n -> + let server = Printf.sprintf {|{"type":"changed","t":%s}|} n in + let client = Printf.sprintf {|{"type":"pull","since":%s}|} n in + ignore (Protocol.decode_server_message server |> require_ok "valid wire cursor"); + ignore (Protocol.decode_client_message client |> require_ok "valid wire cursor")) + [ "0"; "9007199254740991" ]; + List.iter + (fun n -> + let server = Printf.sprintf {|{"type":"changed","t":%s}|} n in + let client = Printf.sprintf {|{"type":"pull","since":%s}|} n in + Alcotest.(check bool) + "server rejects unsafe cursor" + true + (Result.is_error (Protocol.decode_server_message server)); + Alcotest.(check bool) + "client rejects unsafe cursor" + true + (Result.is_error (Protocol.decode_client_message client))) + [ "-1"; "9007199254740992"; "9223372036854775808"; "1.5"; "null"; "\"9\"" ]; + Alcotest.(check bool) + "outbound client rejects unsafe cursor" + true + (Result.is_error + (Protocol.encode_client_message + (Client.Pull { since = Some (server_cursor_of_int 9007199254740992) }))) +;; + let scenarios = [ Alcotest.test_case + "numeric cursor wire safe integer range" + `Quick + test_cursor_wire_range + ; Alcotest.test_case "all client messages round trip" `Quick test_all_client_messages_round_trip diff --git a/logseq_sync/test/transport_contract.ml b/logseq_sync/test/transport_contract.ml index b74dfbe..bc3c836 100644 --- a/logseq_sync/test/transport_contract.ml +++ b/logseq_sync/test/transport_contract.ml @@ -277,8 +277,7 @@ let websocket_start ?(checkpoint = 0) origin = (Logseq_overlay_db.Types.sync_token_of_string "sync-token:v1:transport" |> Result.get_ok) ~checkpoint: - (Logseq_overlay_db.Types.Server_cursor.of_string - ("server-cursor:v1:" ^ string_of_int checkpoint) + (Logseq_overlay_db.Types.Server_cursor.of_int64 (Int64.of_int checkpoint) |> Result.get_ok) ~submissions:[] in @@ -1260,7 +1259,7 @@ let test_output_admission () = let since = if index = 128 then 2000 else index in let expected = Logseq_sync_pure_reducer.Sync_protocol.encode_client_message - (Pull { since = Some since }) + (Pull { since = Some (Core_contract.server_cursor_of_int since) }) |> Result.get_ok |> hex in diff --git a/test/application_view_test.ml b/test/application_view_test.ml index 098bc7a..a13446e 100644 --- a/test/application_view_test.ml +++ b/test/application_view_test.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + module Service = Logseq_db_worker_lui.Logseq_db_worker_lui_service module Wire_nodes = Set.Make (Int) @@ -547,7 +551,7 @@ let run_favorites_native_visibility { sync_phase = Current ; catalog = [] ; selected_graph = Some (uuid 900) - ; applied_server_t = Some 0 + ; applied_server_t = Some (server_cursor_of_int 0) ; timeline_presentation_pending = false ; startup = { authenticated = true @@ -2695,7 +2699,7 @@ let test_reference_outer_terminals cancel = { sync_phase = Current ; catalog = [] ; selected_graph = Some (uuid 900) - ; applied_server_t = Some 0 + ; applied_server_t = Some (server_cursor_of_int 0) ; timeline_presentation_pending = false ; startup = { authenticated = true diff --git a/test/logseq_overlay_db_boundary_test.ml b/test/logseq_overlay_db_boundary_test.ml index c64a4b6..6d6ac4c 100644 --- a/test/logseq_overlay_db_boundary_test.ml +++ b/test/logseq_overlay_db_boundary_test.ml @@ -90,8 +90,8 @@ let test_package_metadata root = "logseq_overlay_db/spec/dune" [ "(name logseq_overlay_db)" ; "(public_name logseq_overlay_db)" - ; "(modules types database)" - ; "(virtual_modules types database)" + ; "(modules server_cursor types database)" + ; "(virtual_modules server_cursor types database)" ; "(default_implementation logseq_overlay_db_impl)" ; "(libraries eio logseq_db_types)" ]; diff --git a/test/source_boundary_test.ml b/test/source_boundary_test.ml index 6ae07e7..af1256d 100644 --- a/test/source_boundary_test.ml +++ b/test/source_boundary_test.ml @@ -551,7 +551,7 @@ let test_standalone_sync_protocol_boundary root = require_text root "logseq_sync/spec/pure_reducer/sync_protocol.mli" - [ "type cursor = int" + [ "type cursor = Logseq_db_types.Server_cursor.t" ; "type checksum = string" ; "module Client : sig" ; "module Server : sig" @@ -583,7 +583,7 @@ let test_standalone_sync_protocol_boundary root = require_text root "logseq_sync/lib/pure_reducer/sync_protocol.ml" - [ "type cursor = int" + [ "type cursor = Logseq_db_types.Server_cursor.t" ; "type checksum = string" ; "module Client = struct" ; "module Server = struct" diff --git a/test/startup_test.ml b/test/startup_test.ml index f8ecd7f..4a2a665 100644 --- a/test/startup_test.ml +++ b/test/startup_test.ml @@ -1,3 +1,7 @@ +let server_cursor_of_int value = + Logseq_db_types.Server_cursor.of_int64 (Int64.of_int value) |> Result.get_ok +;; + module Graph_service = Logseq_db_worker_lui.Logseq_db_worker_lui_service let fail format = Printf.ksprintf failwith format @@ -91,7 +95,7 @@ let startup_snapshot { sync_phase ; catalog = [] ; selected_graph = Some graph_id - ; applied_server_t = Some 11 + ; applied_server_t = Some (server_cursor_of_int 11) ; timeline_presentation_pending ; startup = { authenticated