fix: preserve restored sessions and pending media reads

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

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,17 +936,34 @@ 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
   then retire_coordinated_resources t;
   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
 ;;
 

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 =

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

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

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,19 +1371,26 @@ 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
             | Some id when id<>registration -> session.emit_coordination
               (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

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