diff --git a/app/application.ml b/app/application.ml index 6e4136a..035386a 100644 --- a/app/application.ml +++ b/app/application.ml @@ -3986,6 +3986,10 @@ let start |_->false); coordinated_media_read := (fun ((Journal_media_runtime.Read request) as original)-> match current_owner(),ensure_coordination() with + | Some _,Ok host when Coordination.bootstrap(Coordination_host.view host)=Wc.Awaiting_baseline-> + (* The runtime keeps this read queued. A binding/baseline event pumps it; + temporary admission closure is not a failed media load. *) + false | Some owner,Ok host-> (match Coordination_host.plan_query_when_available host ~owner ~maximum_items:request.maximum_items @@ -4031,6 +4035,7 @@ let start Effect.ignore)) in coordinated_asset_read := (fun (Journal_asset_runtime.Read request) -> match current_owner(),ensure_coordination() with + | Some _,Ok host when Coordination.bootstrap(Coordination_host.view host)=Wc.Awaiting_baseline->false | Some owner,Ok host -> (match Coordination_host.plan_query_when_available host ~owner ~maximum_items:request.maximum_items diff --git a/app/journal_asset_runtime.ml b/app/journal_asset_runtime.ml index 9055488..966baff 100644 --- a/app/journal_asset_runtime.ml +++ b/app/journal_asset_runtime.ml @@ -404,10 +404,10 @@ let dispatch t event = ;; let refresh t ~graph_generation ~today ~settings = - t.resume_configuration<-None; let configuration = graph_generation, today, settings in if t.configuration <> Some configuration then ( + t.resume_configuration<-None; retire_policy_reads t; Hashtbl.clear t.pending; if t.generation <> Some graph_generation @@ -936,9 +936,22 @@ let shutdown t = let retire_coordinated_resources t = if t.resource_binding<>None then ( - let configuration=t.configuration in - shutdown t; - t.resume_configuration<-configuration) + t.resource_binding<-None; + t.resume_configuration<-t.configuration; + (* Keep the scan/dependency tickets that the coordinator will replan. Only + physical demand registrations belong to the retiring binding. *) + let consumers=ref(Hashtbl.to_seq_keys t.bound_consumers |> List.of_seq) in + let reads=Queue.create() and cleanup=Queue.create() in + Queue.iter(function + |Policy.Demand{consumer;_}->consumers:=consumer::!consumers + |Policy.Read _ as instruction->Queue.add instruction reads + |Policy.Release _ as instruction->Queue.add instruction cleanup)t.queued; + let consumers=List.sort_uniq String.compare !consumers in + List.iter(fun consumer->Queue.add(Policy.Release consumer)cleanup)consumers; + Queue.transfer reads cleanup; + t.queued<-cleanup; + List.iter(fun consumer->dispatch t (Policy.Backpressure consumer))consumers; + pump t) ;; let bind_coordinated_resources t binding = if Option.exists(fun previous->not(Worker.equal_binding previous binding))t.resource_binding @@ -946,7 +959,11 @@ let bind_coordinated_resources t binding = let resume=t.resume_configuration in t.resource_binding<-Some binding;t.resume_configuration<-None; match resume with - |Some(graph_generation,today,settings)->refresh t ~graph_generation ~today ~settings + |Some _-> + (* Resync preserves in-flight read ownership, but discards any sealed scan + continuation before rebuilding demand against the replacement binding. *) + dispatch t Policy.Resync; + dispatch t Policy.Capacity_available |None->pump t ;; diff --git a/app/journal_media_runtime.ml b/app/journal_media_runtime.ml index ce096df..792301a 100644 --- a/app/journal_media_runtime.ml +++ b/app/journal_media_runtime.ml @@ -1519,17 +1519,22 @@ let retire_coordinated_resources t = if t.resource_binding<>None then ( t.resource_binding<-None; t.queued<-List.filter(fun outgoing->match outgoing.request with - |Release_preview _|Release_consumer _->true - |Demand _|Acquire _|Retry_asset _|Perform_read _->retire_outgoing t outgoing;false)t.queued; - Hashtbl.iter(fun _ id->t.read_retired id)t.fact_requests; - Hashtbl.clear t.fact_requests;Hashtbl.clear t.facts;Hashtbl.clear t.pages; - Hashtbl.clear t.page_barriers;Hashtbl.clear t.sources;Hashtbl.clear t.barriers; + |Release_preview _|Release_consumer _|Perform_read _->true + |Demand _|Acquire _|Retry_asset _->retire_outgoing t outgoing;false)t.queued; + (* Binding retirement invalidates resource capabilities, not the graph's + logical read owners. The coordinator retains and replans those reads; + clearing their slots here would strand shared dependency consumers. *) retire_coordinated_previews t; Hashtbl.iter(fun _ g-> - Option.iter t.read_retired g.pending;g.pending<-None;g.cursor<-None; - g.membership_changed<-true;g.dependency_failed<-false;g.listing_failed<-false;g.error<-None; + g.cursor<-None; + if g.pending=None then g.membership_changed<-true; + g.dependency_failed<-false;g.listing_failed<-false;g.error<-None; List.iter(fun c->dispatch t g c P.Hide)g.controllers; notify t g)t.groups; + (* Cleanup must not wait for read admission or a successful fresh baseline. *) + let cleanup,waiting=List.partition(fun outgoing->match outgoing.request with + |Release_preview _|Release_consumer _->true|_->false)t.queued in + t.queued<-cleanup @ waiting; pump t) ;; let resume_coordinated_previews t = diff --git a/logseq_db_worker/lui/journal_worker.ml b/logseq_db_worker/lui/journal_worker.ml index 158108b..d62fdf2 100644 --- a/logseq_db_worker/lui/journal_worker.ml +++ b/logseq_db_worker/lui/journal_worker.ml @@ -92,6 +92,7 @@ module Session_context = struct ; emit : topic:ID.Worker.push_topic -> 'push -> unit ; fork_daemon : name:string -> (unit -> unit) -> unit ; issue_control : topic:int -> control_ticket option + ; latest_control_version : topic:int -> int option } let switch t = t.switch @@ -102,6 +103,7 @@ module Session_context = struct let emit t = t.emit let fork_daemon t = t.fork_daemon let issue_control t = t.issue_control + let latest_control_version t = t.latest_control_version end module Request_context = struct @@ -1075,6 +1077,9 @@ let run_direct_session ; emit = emit_push ; fork_daemon = start_daemon ; issue_control = issue_control_ticket client + ; latest_control_version = (fun ~topic -> with_control_lock client (fun () -> + let slot = control_slot client topic in + Option.map (fun _ -> slot.version) slot.invalidation_target)) } in let init_result, resolve_init = Eio.Promise.create () in diff --git a/logseq_db_worker/lui/journal_worker.mli b/logseq_db_worker/lui/journal_worker.mli index 25d0454..f19ee54 100644 --- a/logseq_db_worker/lui/journal_worker.mli +++ b/logseq_db_worker/lui/journal_worker.mli @@ -1,4 +1,23 @@ -(** Typed domain-0 client and Worker Domain service contract. *) +(** {1 TLDR} + + Typed domain-0 client and Worker Domain service contract. Services own their + session state; clients enqueue requests and receive correlated outcomes. + [send] reports admission, not completion. Latest controls can supersede + queued state independently of ordinary request admission. + + {1 Minimal usage example} + + Given a ready [client] and service [request], retain the admitted ID and + match it against later {!event} responses: + {[ + match send client request with + | Accepted request_id -> track request_id + | Full | Not_ready | Stopping -> defer_request request + ]} + [track] and [defer_request] are caller-supplied callbacks. Cancellation does + not imply rollback of an already accepted durable operation. + + {1 Construction and validation} *) type mono_clock = Eio.Time.Mono.ty Eio.Resource.t type net = [ `Generic ] Eio.Net.ty Eio.Resource.t @@ -41,6 +60,14 @@ module Session_context : sig the previous obligation before issuing a replacement. Returns [None] during stop or while an invalidating latest update is pending. *) val issue_control : 'push t -> topic:int -> control_ticket option + + (** Number of distinct latest-state keys admitted on this session's topic, + including superseded/coalesced updates and updates whose ordinary request + was full or cancelled. Identical consecutive keys do not advance it. + [None] means no latest-state admission; reply topics also return [None]. + Read atomically under the control mutex; this is not a handler count and + may already include an update newer than the executing handler. *) + val latest_control_version : 'push t -> topic:int -> int option end module Request_context : sig 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 ac417ff..bfc678b 100644 --- a/logseq_db_worker/lui/logseq_db_worker_lui_service.ml +++ b/logseq_db_worker/lui/logseq_db_worker_lui_service.ml @@ -1021,6 +1021,8 @@ let create ~(dependencies : dependencies) = ; auth_enqueued_sequence : int ref ; auth_wake : Eio.Condition.t ; mutable coordination_registration : int64 option + ; mutable coordination_account : Sync.account_scope option + ; first_authentication : unit -> bool ; coordination_limits : Worker_pure_data.capability_limits ; emit_observation : observation_signal -> unit ; emit_coordination : coordination_update -> unit @@ -1062,10 +1064,23 @@ let create ~(dependencies : dependencies) = request.request "host rejected request" | Client_command (Reconcile_authenticated_user { user_id }) -> - Option.iter (fun id -> session.Session.emit_coordination - (Coordination_revoked(id,"Authentication authority changed"))) session.coordination_registration; - session.coordination_registration<-None; - Db.revoke_coordination session.worker; + (* Confirming the restored account does not change graph authority. + Only the first admitted identity can make this transition: a first + executed A may hide a coalesced B -> A or signed-out -> A. *) + let confirms_restored_account = + not !(session.Session.auth_seen) && session.first_authentication () + && match user_id, session.coordination_registration, session.coordination_account with + | Some user, Some registration, Some account -> + String.equal user account.user_id + && Db.coordination_current session.worker ~registration + | _ -> false + in + if not confirms_restored_account then ( + Option.iter (fun id -> session.Session.emit_coordination + (Coordination_revoked(id,"Authentication authority changed"))) session.coordination_registration; + session.coordination_registration<-None; + session.coordination_account<-None; + Db.revoke_coordination session.worker); session.Session.authenticated_user := user_id; (* Every admitted changed key revoked the old reply ticket, including A -> B -> A coalescing. Preserve that invalidation even when only @@ -1207,6 +1222,9 @@ let create ~(dependencies : dependencies) = ; auth_enqueued_sequence ; auth_wake ; coordination_registration=None + ; coordination_account=None + ; first_authentication=(fun () -> + Journal_worker.Session_context.latest_control_version context ~topic:1 = Some 1) ; coordination_limits=(let limits=Overlay.dependency_limits dependencies.overlay in {response_budget_bytes=min limits.response_budget_bytes (min config.response_budget_bytes Worker_pure_data.maximum_response_bytes); @@ -1353,7 +1371,10 @@ let create ~(dependencies : dependencies) = let baseline snapshot={registration;baseline=snapshot;limits=session.Session.coordination_limits; authority_user=(if !(session.auth_seen) then Some !(session.authenticated_user) else None)} in (match Db.register_coordination session.worker ~registration - ~notify:(fun snapshot -> session.emit_coordination (Coordination_snapshot(baseline snapshot))) with + ~notify:(fun snapshot -> + if session.coordination_registration=Some registration then + session.coordination_account<-snapshot.state.sync.account; + session.emit_coordination (Coordination_snapshot(baseline snapshot))) with | Error message -> Error message | Ok snapshot -> (match session.coordination_registration with @@ -1361,11 +1382,15 @@ let create ~(dependencies : dependencies) = (Coordination_revoked(id,"Coordination binding replaced")) | _ -> ()); session.coordination_registration<-Some registration; + session.coordination_account<-snapshot.state.sync.account; Ok(Coordination_registered(baseline snapshot))) | Release_coordination registration -> Result.map(fun()-> (match session.coordination_registration with - | Some id when id<=registration -> session.coordination_registration<-None | _ -> ()); + | Some id when id<=registration -> + session.coordination_registration<-None; + session.coordination_account<-None + | _ -> ()); Coordination_released registration)(Db.release_coordination session.worker ~registration) | Coordination_query {registration;owner;request_id;read} -> let Read.Read_query(budget,query)=read in diff --git a/test/journal_asset_policy_test.ml b/test/journal_asset_policy_test.ml index f13a640..f9f1813 100644 --- a/test/journal_asset_policy_test.ml +++ b/test/journal_asset_policy_test.ml @@ -416,6 +416,83 @@ let runtime_same_configuration () = ;; +let runtime_binding_retains_reads () = + let sent = Queue.create () in + let retired = ref [] in + let r = Runtime.create + ~read:(fun request -> Queue.add (Read request) sent; true) + ~read_retired:(fun id -> retired := id :: !retired) + ~demand:(demand_port sent) ~release_consumer:(release_port sent) + ~changed:(fun _ _ _ -> ()) () in + let module I = (val resources.Resources.issuer : Worker_effect_authority.ISSUER) in + Runtime.bind_coordinated_resources r I.binding; + Runtime.refresh r ~graph_generation:1 ~today:20260301 ~settings:P.default_settings; + let requests = Queue.to_seq sent |> Seq.filter_map(function Read request->Some request|_->None) |> List.of_seq in + Queue.clear sent; + Runtime.retire_coordinated_resources r; + List.iter(fun request -> check (Runtime.owns_read r (request_id request)) + "resource handover lost an offline scan read") requests; + Runtime.bind_coordinated_resources r I.binding; + check (Queue.is_empty sent) "binding handover duplicated pending offline scans"; + check (!retired = []) "binding handover retired shared scan consumers"; + Runtime.shutdown r; + List.iter(fun request -> check (not(Runtime.owns_read r (request_id request))) + "account revocation kept offline scan ownership") requests +;; + +let runtime_binding_restarts_waiting_demand accepted () = + let sent=Queue.create() and released=ref 0 in + let allow_reads=ref true in + let r=Runtime.create + ~read:(fun request->if !allow_reads then (Queue.add(Read request)sent;true)else false) + ~demand:(fun ~graph_generation ~name priority assets-> + if accepted then demand_port sent ~graph_generation ~name priority assets else None) + ~release_consumer:(fun registration->incr released;release_port sent registration) + ~changed:(fun _ _ _->()) () in + let module I=(val resources.Resources.issuer:Worker_effect_authority.ISSUER) in + Runtime.bind_coordinated_resources r I.binding; + let settings=P.settings ~recent_days:0 |> Result.get_ok in + Runtime.refresh r ~graph_generation:1 ~today:20260301 ~settings; + let take()=match Queue.take sent with Read request->request|_->failwith "expected read" in + respond r (take()) (fun(Read.Read_query(budget,query))-> + let module F=Read.For_worker in + match query with + |Read.Favorites _->F.answer budget query(Ok(F.favorites ~favorites_page:None + ~generation:"g" ~projection_revision:"r" ~next:None + ~items:[{membership_uuid=uuid 10;membership_order="a";membership_revision="r"; + target=Worker_pure_data.V2_favorite_page{uuid=uuid 1;title="Page";revision="r"}}])) + |_->failwith "expected favorites"); + while not(Queue.is_empty sent) do + match Queue.take sent with + |Command _->() + |Read request->respond r request (fun(Read.Read_query(budget,query))-> + let module F=Read.For_worker in match query with + |Read.Assets _->F.answer budget query(Ok(F.asset_listing ~generation:"g" + ~projection_revision:"r" ~items:[asset 50] ~next:None)) + |Read.Page{uuid;_}->F.answer budget query(Ok(Worker_pure_data.V2_present_page + {page={G.uuid;name="page";title="Page";kind=Ordinary_page;created_at_ms=0L; + updated_at_ms=0L;tags=[];properties=[];recycled=false};revision="r"})) + |_->failwith "expected assets or their page dependency") + done; + Queue.clear sent; + allow_reads:=false; + Runtime.retire_coordinated_resources r; + if accepted then check(!released=1) "retired demand cleanup waited behind a read"; + (* Ordinary repeated model synchronization must not erase a pending rebind. *) + Runtime.refresh r ~graph_generation:1 ~today:20260301 ~settings; + allow_reads:=true; + Runtime.bind_coordinated_resources r I.binding; + check(Queue.to_seq sent |> Seq.exists(function + |Read(Runtime.Read{query=W.Favorites None;_})->true|_->false)) + "retiring an unacknowledged demand stranded its offline scan" +;; + +let () = + runtime_binding_retains_reads (); + runtime_binding_restarts_waiting_demand false (); + runtime_binding_restarts_waiting_demand true () + + let completed_favorites () = let s, ins = refresh P.empty in let s, ins = P.step s (Roots_loaded (read Favorites ins, [ uuid 1 ], None)) in diff --git a/test/journal_media_runtime_test.ml b/test/journal_media_runtime_test.ml index aa0b4c7..fbf5575 100644 --- a/test/journal_media_runtime_test.ml +++ b/test/journal_media_runtime_test.ml @@ -513,9 +513,104 @@ let run_cases cases = check (!failed = 0) (Printf.sprintf "%d media dependency cases failed" !failed) ;; +(* Resource retirement is owned by this adapter, outside the coordinator reducer. + Its expected-owner read tickets already survive a binding handover. *) +let test_binding_retains_media_reads () = + List.iter (fun accepted -> + let r,q,views,admitted = fixture () in + let module I = (val resource_source.Resources.issuer : Worker_effect_authority.ISSUER) in + R.bind_coordinated_resources r I.binding; + admitted := accepted; + let root = G.Uuid.to_string (uuid 1) in + R.root_visible ~owner:"timeline" r ~root true; + R.root_visible ~owner:"detail" r ~root true; + let original = if accepted then Some(take_read q) else None in + admitted := false; + R.retire_coordinated_resources r; + Option.iter (fun request -> check (R.owns_read r (R.read_request_id request)) + "resource handover retired a live media read") original; + R.root_visible ~owner:"timeline" r ~root false; + admitted := true; + R.bind_coordinated_resources r I.binding; + let request = match original with Some request -> request | None -> take_read q in + check (Queue.is_empty q) "shared visible owners duplicated a retained media read"; + reply r request []; + check (Queue.is_empty q) "retained media completion unnecessarily restarted its listing"; + check ((Hashtbl.find views root).error = None) "binding handover published a media error"; + R.reset r ~graph_generation:(Some 2); + check (not(R.owns_read r (R.read_request_id request))) "graph replacement kept old media ownership" + ) [false;true] +;; + +let test_binding_cleanup_precedes_waiting_reads () = + let sent=Queue.create() and allow_reads=ref true and released=ref 0 and views=Hashtbl.create 2 in + let ports=resource_ports sent (fun()->true) in + let r=R.create + ~read:(fun request->if !allow_reads then(Queue.add(Read request)sent;true)else false) + ~demand:ports.demand ~acquire:ports.acquire ~release_file:ports.release_file + ~release_consumer:(fun registration->incr released;ports.release_consumer registration) + ~retry_asset:ports.retry_asset ~changed:(fun root view->Hashtbl.replace views root view) () in + R.reset r ~graph_generation:(Some 1); + let module I=(val resource_source.Resources.issuer:Worker_effect_authority.ISSUER) in + R.bind_coordinated_resources r I.binding; + let root=G.Uuid.to_string(uuid 1) in + let asset=A.create ~uuid:(uuid 2) + ~source:(Managed(Some(A.version ~checksum:(String.make 64 'a') ~file_type:"png" |> Result.get_ok))) + ~current_checksum:None ~size:None ~dimensions:None |> Result.get_ok in + loaded r sent root [asset]; + let token=(List.hd(Hashtbl.find views root).items).token in + R.asset_visible r ~root ~asset:token true; + Queue.clear sent; + R.notice r (Resources.notice resource_source(S.Asset_availability + {consumer=token;asset=asset.uuid;descriptor=asset;availability=Ready "cached"})); + let acquisition=take_acquire sent in + file_reply r acquisition "rebind-held" "/cache/rebind-held"; + Queue.clear sent; + allow_reads:=false; + R.root_visible r ~root:(G.Uuid.to_string(uuid 3)) true; + R.retire_coordinated_resources r; + check(!released>0) "waiting read blocked retired demand cleanup"; + check(commands sent |> Seq.exists(fun command->released_handle command=Some "rebind-held")) + "waiting read blocked retired file cleanup"; + check(list_reads sent=0) "waiting read was admitted while rebinding" +;; + +let test_binding_retains_shared_facts () = + let r,q,views,admitted = fixture () in + let module I = (val resource_source.Resources.issuer : Worker_effect_authority.ISSUER) in + R.bind_coordinated_resources r I.binding; + observe r (raw_block 1 60); + observe r (raw_block 2 60); + loaded r q (G.Uuid.to_string(uuid 1)) []; + loaded r q (G.Uuid.to_string(uuid 2)) []; + change r [80]; + let original = take_read q in + check (Queue.is_empty q) "two roots did not share the missing ancestor read"; + admitted := false; + R.retire_coordinated_resources r; + check (R.owns_read r (R.read_request_id original)) "binding handover stranded shared ancestor consumers"; + R.root_visible r ~root:(G.Uuid.to_string(uuid 1)) false; + check (R.owns_read r (R.read_request_id original)) "one hidden root canceled another root's shared read"; + Queue.add(Read original)q; + admitted:=true; + R.bind_coordinated_resources r I.binding; + reply_block r q (raw_block 60 900); + check(not(R.owns_read r (R.read_request_id original))) "retained ancestor completion was not consumed"; + check((Hashtbl.find views (G.Uuid.to_string(uuid 2))).error=None) + "remaining shared consumer failed after a binding handover"; + let pending=take_metadata q in + R.reset r ~graph_generation:None; + check (not(R.owns_read r (R.read_request_id pending))) "account revocation retained media read"; + reply r pending []; + check(Queue.is_empty q) "late read after account revocation restarted media work" +;; + let () = run_cases - [ ("queued coordinated read retires on Hide",coordinated_read_retirement false) + [ ("binding retains media reads",test_binding_retains_media_reads) + ; ("binding cleanup precedes waiting reads",test_binding_cleanup_precedes_waiting_reads) + ; ("binding retains shared facts",test_binding_retains_shared_facts) + ; ("queued coordinated read retires on Hide",coordinated_read_retirement false) ; ("accepted coordinated read retires on reset",coordinated_read_retirement true) ; ( "stale initial holder facts refill without poisoning the root" , fun () -> diff --git a/test/worker_coordination_runtime_test.ml b/test/worker_coordination_runtime_test.ml index c233968..20bb6fe 100644 --- a/test/worker_coordination_runtime_test.ml +++ b/test/worker_coordination_runtime_test.ml @@ -583,7 +583,7 @@ let test_application_current_close ~replacement () = press "accessibility-label" "Favorites"; press "text" "Switch graph"; wait "native identity reaches the deliberately blocked token request" (fun()-> !token_requested); - wait "same-user authentication rebind completes before the close action" (fun()-> !registrations>=2); + require (!registrations=1) "same-user authentication preserves the original app registration"; if Option.is_none(find "accessibility-identifier" "graph-picker-list")then press "text" "Switch graph"; List.iter(fun graph_id-> press "accessibility-identifier" ("graph-picker:" ^ W.Uuid.to_string graph_id); @@ -1951,7 +1951,7 @@ let test_client_to_typed_read () = "late cancelled read revived a forgotten consumer") ;; -let test_read_completion_rechecks_graph_owner () = +let test_read_completion_rechecks_graph_owner ?(change=`Graph) () = let armed = Atomic.make false and entered = Atomic.make false and release = Atomic.make false in @@ -1969,7 +1969,7 @@ let test_read_completion_rechecks_graph_owner () = ~finally:(fun () -> Atomic.set release true) (fun () -> let responses = Hashtbl.create 16 - and baseline = ref None in + and baseline = ref None and authority_revoked = ref false in Mailbox.on_event client (function | Mailbox.Response { request_id; outcome; _ } -> Hashtbl.replace responses request_id outcome @@ -1977,7 +1977,7 @@ let test_read_completion_rechecks_graph_owner () = List.iter (function | Service.Coordination_snapshot value -> remember_baseline baseline value - | _ -> ()) + | Coordination_revoked _ -> authority_revoked := true) updates | _ -> ()); let wait stage predicate = @@ -2052,24 +2052,34 @@ let test_read_completion_rechecks_graph_owner () = require (not (Hashtbl.mem responses read)) "read did not wait behind actual writer"; - ignore (send (Service.Client_command Return_to_graph_picker)); - wait "graph picker published" (fun () -> - match !baseline with - | Some value -> value.baseline.state.sync.selected = None - | _ -> false); + (match change with + | `Graph -> + ignore (send (Service.Client_command Return_to_graph_picker)); + wait "graph picker published" (fun () -> + match !baseline with + | Some value -> value.baseline.state.sync.selected = None + | _ -> false) + | `Same | `Changed | `Signout -> + let user_id=match change with `Same->Some "user-1"|`Changed->Some "other-user"|_->None in + let auth=send(Service.Client_command(Reconcile_authenticated_user{user_id})) in + if change=`Same then wait "same identity confirmed" (fun()->Hashtbl.mem responses auth) + else wait "different authority revoked" (fun()-> !authority_revoked)); Atomic.set release true; wait "retired read completion" (fun () -> Hashtbl.mem responses read); require (match Hashtbl.find responses read with - | Failed _ -> true + | Completed(Service.Coordination_queried{answer=Read.Query_answer(_,Read.Graph_info,Ok info);_}) + when change=`Same -> info.graph_uuid=Test_support.graph_uuid + | Completed(Service.Coordination_query_revoked _) | Failed _ when change<>`Same -> true | _ -> false) - "retired graph read completion still crossed Service's authority gate"; + "in-flight read did not respect the actual identity/graph transition"; ( wait "cancelled writer completion" (fun () -> Hashtbl.mem responses writer); require (Hashtbl.find responses writer = Mailbox.Cancelled) "cancelled durable attempt was reported as known rollback"; Atomic.set armed false; + if change=`Graph then ( ignore (send (Service.Client_command (Select_graph Test_support.graph_uuid))); wait "replacement graph open" (fun () -> match !baseline with @@ -2109,7 +2119,7 @@ let test_read_completion_rechecks_graph_owner () = (committed.before <> committed.after) "unknown accepted write lost original commit evidence" - | _ -> failwith "durable unknown did not reconcile across close/reopen"))) + | _ -> failwith "durable unknown did not reconcile across close/reopen")))) ;; let test_host_durable_recovery_confirmation_ack () = @@ -2642,88 +2652,140 @@ let test_dependency_point_reads_use_typed_owner () = wait (fun () -> C.shutdown_state (Host.view host) = C.Stopped)) ;; -(* A real Service authority control can retire an offline binding without - changing its user or physical graph. The old pure event only represented - terminal bootstrap revocation, so it could not express this live handover. *) -let test_live_authority_invalidation_rebinds () = - let module Host = Journal_worker_coordination_host in - with_client (fun client -> - let jobs = Queue.create () in - let host = - Host.create - ~client - ~client_id - ~managed_origin:"https://api.logseq.io" - ~user_id:"user-1" - ~enqueue:(fun f -> Queue.add f jobs) - ~changed:(fun _ -> ()) - |> Result.get_ok - in - Mailbox.on_event client (fun event -> ignore (Host.on_event host event)); - let pump () = - Mailbox.Private.deliver client ~max_events:64; - let count = Queue.length jobs in - for _ = 1 to count do - (Queue.take jobs) () - done - in - let wait predicate = - let deadline = Unix.gettimeofday () +. 6. in - while (not (predicate ())) && Unix.gettimeofday () < deadline do - pump (); - Thread.delay 0.001 - done; - require (predicate ()) "authority handover did not settle" - in - let current () = - Option.bind - (C.projection (Host.view host)) - (fun p -> - List.find_map - (fun r -> if r.W.phase = W.Open then Some r.session else None) - (W.residents p)) - in - ignore - (Mailbox.send - client - (Service.Client_command (Restore_local_account { user_id = "user-1" }))); - wait (fun () -> current () <> None); - let old = Option.get (current ()) in - ignore - (Mailbox.send - client - (Service.Client_command - (Reconcile_authenticated_user { user_id = Some "user-1" }))); - wait (fun () -> - match C.bootstrap (Host.view host), current () with - | W.Binding_revoked _, _ -> true - | _, Some session -> not (W.equal_session old session) - | _ -> false); - let session = - match current () with - | Some value -> value - | None -> - failwith "same-user live authority left the offline binding permanently revoked" - in - require - (not (W.equal_session old session)) - "authority handover reused its revoked registration"; - let budget = - W.read_budget session ~maximum_items:1 ~maximum_bytes:65536 |> Result.get_ok - in - let ticket = - Host.request host session (W.Query (budget, W.Graph_info)) |> Result.get_ok - in - let completed = ref false in - Host.watch host ticket ~f:(function - | W.Answered (Ok _) -> completed := true - | _ -> failwith "replacement binding cannot execute a real read"); - wait (fun () -> !completed); - require - (Host.validate_session host old = Error W.Revoked_session) - "old authority session recovered data admission"; - ignore (Host.shutdown host); - wait (fun () -> C.shutdown_state (Host.view host) = C.Stopped)) +(* The pure Sync reducer already retains same-user generations. Registration + invalidation and coalesced SDK controls belong to the real Service boundary. *) +let test_restored_authentication_session mode () = + with_service_fixture (fun dependencies config -> + let client, startup = Mailbox.Private.prepare + ~runtime_epoch:(Id.Runtime.Epoch.of_int64 9200L) + ~worker_generation:(Id.Worker.Generation.of_int64 1L) + (Service.create ~dependencies) config in + let ready = Atomic.make false and pause = Atomic.make false + and paused = Atomic.make false in + let worker = Domain.spawn (fun () -> + Logseq_db_worker_lui.Journal_worker_eio_backend.run (fun environment -> + Eio.Switch.run (fun session_switch -> + Mailbox.Private.run_session startup ~environment ~session_switch + ~on_startup:(fun result -> require (Result.is_ok result) "auth Service startup"; Atomic.set ready true) + ~on_idle_wait:(fun () -> if Atomic.get pause then ( + Atomic.set paused true; + while Atomic.get pause do Thread.delay 0.001 done; + Atomic.set paused false)) + ~on_yield:(fun () -> ())))) in + Fun.protect + ~finally:(fun () -> Atomic.set pause false; Mailbox.Private.request_stop client; ignore(Domain.join worker)) + (fun () -> + let responses = Hashtbl.create 32 and baseline = ref None and revoked = ref [] in + let pump () = Mailbox.For_testing.drain_events client ~max_events:128 |> List.iter (function + | Mailbox.Response {request_id;outcome;_} -> + require (not(Hashtbl.mem responses request_id)) "auth request completed twice"; + Hashtbl.add responses request_id outcome + | Push {payload=Service.Coordination_updates updates;_} -> + List.iter (function + | Service.Coordination_snapshot value -> remember_baseline baseline value + | Coordination_revoked(id,_) -> revoked := id :: !revoked) updates + | _ -> ()) in + let wait stage predicate = + let deadline = Unix.gettimeofday () +. 6. in + while not(predicate()) && Unix.gettimeofday() id | _ -> failwith "auth request admission" in + let result id = wait "request completion" (fun () -> Hashtbl.mem responses id); Hashtbl.find responses id in + let call request = result(send request) in + let auth user_id = Service.Client_command(Reconcile_authenticated_user{user_id}) in + wait "ready" (fun () -> Atomic.get ready); + (* Register before restoration to require tracking the live baseline, + rather than only its original empty registration snapshot. *) + let registration = 9200L in + (match call(Service.Register_coordination registration) with + | Completed(Service.Coordination_registered value) -> remember_baseline baseline value + | _ -> failwith "auth registration"); + ignore(call(Service.Client_command(Restore_local_account{user_id="user-1"}))); + wait "restored graph" (fun () -> Option.exists (fun value -> + List.exists(fun(_,state)->state.Logseq_db_worker.phase=Graph_open) + value.Service.baseline.state.residents) !baseline); + let old = Option.get !baseline in + let owner = Option.get old.baseline.state.sync.selected in + let source = observation_read client !baseline owner registration Read.Graph_info in + let id n=uuid(Printf.sprintf "9c000000-0000-4000-8000-%012d" (9200+n)) in + let query n=Service.Coordination_query{registration;owner;request_id=id n;read=source} in + let observe=Service.Coordination_observe{registration;owner;observation_id=id 10;request_id=id 11;read=source} in + let successful = function + | Mailbox.Completed(Service.Coordination_queried{answer=Read.Query_answer(_,Read.Graph_info,Ok info);_}) -> + info.graph_uuid=Test_support.graph_uuid + | _ -> false in + require(successful(call(query 1))) "restored session cannot read"; + require(match call observe with Completed(Service.Coordination_observed{epoch=0L;_})->true|_->false) + "restored observation not established"; + if mode="same-online" then + ignore(call(Service.Client_command(Request_sign_in owner.startup))); + if mode="verified-aba" then ignore(call(auth(Some "user-1"))); + Atomic.set pause true; + ignore(send Service.Get_graph_state); + wait "scheduler paused" (fun () -> Atomic.get paused); + let pending=send(query 2) in + let keys=match mode with + | "same" | "same-online" | "full-same" -> [Some "user-1"] + | "duplicate" -> [Some "user-1";Some "user-1"] + | "changed" -> [Some "other-user"] + | "signout" | "full-signout" | "cancelled-signout" -> [None] + | "first-aba" | "verified-aba" -> [Some "other-user";Some "user-1"] + | "first-signout-a" -> [None;Some "user-1"] + | _ -> failwith "unknown auth session case" in + if String.starts_with ~prefix:"full-" mode then ( + let full=ref false in + for _=1 to 128 do if not !full then + match Mailbox.send client Service.Get_graph_state with + | Full -> full:=true | Accepted _ -> () | _ -> failwith "queue stopped" done; + require !full "auth fixture did not fill response reservations"); + let confirmations=List.filter_map(fun user -> match Mailbox.send client(auth user) with + | Accepted id -> + if mode="cancelled-signout" then Mailbox.cancel client ~request_id:id; + Some id + | Full when String.starts_with ~prefix:"full-" mode -> None + | _ -> failwith "auth control admission") keys in + Atomic.set pause false; + let outcome=result pending in + List.iter(fun id -> ignore(result id)) confirmations; + (* Data completion is after the pending Latest control even on Full. *) + let preserves=List.mem mode ["same";"same-online";"duplicate";"full-same"] in + if preserves then ( + require(successful outcome) "same-user confirmation revoked an outstanding read"; + require(not(List.mem registration !revoked)) "same-user confirmation published revocation"; + (match call(Service.Register_coordination registration) with + | Completed(Service.Coordination_registered current) -> + require(current.baseline.state.sync.account=old.baseline.state.sync.account) + "same-user confirmation replaced account generations"; + require(current.baseline.state.sync.selected=old.baseline.state.sync.selected) + "same-user confirmation replaced graph/presentation ownership" + | _ -> failwith "same-user confirmation retired the original registration"); + require(successful(call(query 3))) "original session cannot read after confirmation"; + require(match call observe with Completed(Service.Coordination_observed{epoch=0L;_})->true|_->false) + "same-user confirmation retired the original observation"; + (* A genuine new graph lifetime still rejects the old capability. *) + ignore(call(Service.Client_command Return_to_graph_picker)); + wait "graph closed" (fun () -> Option.exists(fun value->value.Service.baseline.state.sync.selected=None)!baseline); + require(not(successful(call(query 4)))) "same-user exemption revived a closed graph"; + ignore(call(Service.Client_command(Select_graph Test_support.graph_uuid))); + wait "graph reopened" (fun () -> Option.exists(fun value -> + List.exists(fun((scope:Sync.graph_scope),state)->scope.graph_generation<>owner.graph_generation + && state.Logseq_db_worker.phase=Graph_open)value.Service.baseline.state.residents)!baseline); + require(not(successful(call(query 5)))) "same-user exemption crossed a new graph lifetime") + else ( + require(not(successful outcome)) "changed authority completed an old outstanding read"; + wait "binding revoked" (fun () -> List.mem registration !revoked); + require(not(successful(call(query 3)))) "changed authority revived old query admission"; + require(match call(Service.Register_coordination registration) with Failed _->true|_->false) + "retired registration became live again"))) +;; + +let test_restored_authentication_sessions () = + List.iter (fun mode -> test_restored_authentication_session mode ()) + ["same";"same-online";"duplicate";"changed";"signout";"first-aba";"first-signout-a"; + "verified-aba";"full-same";"full-signout";"cancelled-signout"]; + List.iter (fun change -> test_read_completion_rechecks_graph_owner ~change ()) + [`Same;`Changed;`Signout] ;; let test_changefeed_business_waits_for_ui_application () = @@ -4510,7 +4572,7 @@ let test_retired_query_has_correlated_response () = let budget=W.read_budget session ~maximum_items:1 ~maximum_bytes:65536 |> Result.get_ok in let id=W.request_id(uuid "9c000000-0000-4000-8000-000000009901") in let request=I.request session ~request_id:id(W.Query(budget,W.Graph_info)) |> Result.get_ok in - ignore(send(Service.Client_command(Reconcile_authenticated_user{user_id=Some "user-1"}))); + ignore(send(Service.Client_command(Reconcile_authenticated_user{user_id=None}))); let response=match send request with |Mailbox.Completed response->response |_->failwith "retired query used an uncorrelated transport failure instead of an owned authority result" in @@ -4527,6 +4589,10 @@ let test_retired_query_has_correlated_response () = let () = test_cleanup_transport_causes(); if Array.mem "--cleanup-causes-only" Sys.argv then exit 0; + if Array.mem "--auth-session-only" Sys.argv then ( + test_restored_authentication_sessions(); + test_retired_query_has_correlated_response(); + print_endline "Restored same-user sessions and real authority retirement passed";exit 0); if Array.mem "--read-revocation-only" Sys.argv then (test_retired_query_has_correlated_response();exit 0); if Array.mem "--application-close-only" Sys.argv then (test_application_current_close ~replacement:false();test_application_current_close ~replacement:true();exit 0); test_application_current_close ~replacement:false(); @@ -4568,7 +4634,7 @@ let () = test_visible_media_listing_uses_typed_reads (); test_offline_asset_scans_use_typed_reads (); test_dependency_point_reads_use_typed_owner (); - test_live_authority_invalidation_rebinds (); + test_restored_authentication_sessions (); test_retired_query_has_correlated_response (); test_service_changefeed_has_one_owner (); test_service_unallocated_observation_cleanup_after_close ();