From a5c6a50bc106afff05c3694f2e347065780ee8c0 Mon Sep 17 00:00:00 2001 From: rcmerci Date: Fri, 16 Oct 2026 14:48:50 +0800 Subject: [PATCH 1/4] Show captured attachments from local files before upload --- app/application.ml | 11 +- app/journal_media.ml | 7 +- app/journal_media_runtime.ml | 57 ++++- .../bugfix/2026-10-16-capture-local-assets.md | 51 +++++ .../lib/effect_runner/effect_runner.ml | 168 +++++++++++++-- logseq_db_worker/test/test_effect_runner.ml | 53 ++++- .../lib/effect_runner/effect_runner.ml | 20 ++ .../lib/pure_reducer/asset_transfer.ml | 170 ++++++++++----- logseq_sync/lib/pure_reducer/core.ml | 7 + .../spec/pure_reducer/asset_transfer.mli | 29 ++- logseq_sync/spec/pure_reducer/core.mli | 5 + logseq_sync/test/runner_contract.ml | 80 +++++++ logseq_sync/test/test_asset_transfer.ml | 52 ++++- test/application_view_test.ml | 33 +-- test/journal_media_runtime_test.ml | 197 ++++++++++++++++++ test/journal_media_test.ml | 13 ++ 16 files changed, 842 insertions(+), 111 deletions(-) create mode 100644 docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md diff --git a/app/application.ml b/app/application.ml index 006cf2c1..4da5c7e0 100644 --- a/app/application.ml +++ b/app/application.ml @@ -4737,7 +4737,7 @@ let start let completion = let current = match pending with - | Some (generation, _, owner) -> + | Some (generation, operation, owner) -> let snapshot = !state_ref in generation = snapshot.graph_state.generation && @@ -4752,11 +4752,14 @@ let start then Some routes else None)) in - (match result, Option.bind routes Journal_routes.detail with - | Ok receipt, Some detail -> + (match result, owner, Option.bind routes Journal_routes.detail with + | Ok receipt, None, _ -> + Logseq_db_types.Graph_types.Uuid.to_string receipt.operation = operation + && receipt.graph_generation = generation + | Ok receipt, Some _, Some detail -> Journal_model.id (Journal_detail.root detail) = Logseq_db_types.Graph_types.Uuid.to_string receipt.target - | Error _, _ -> true + | Error _, _, _ -> true | _ -> false) | None -> false in diff --git a/app/journal_media.ml b/app/journal_media.ml index b82d379b..d029c4a4 100644 --- a/app/journal_media.ml +++ b/app/journal_media.ml @@ -247,7 +247,12 @@ let step t = function | Some old when old.graph_generation = graph_generation && old.consumer = consumer - && old.asset.uuid = asset.uuid -> { t with selection = Some selected }, [] + && old.asset.uuid = asset.uuid + && (match t.status with + | File _ -> true + | _ -> old.asset.source = asset.source + && old.asset.current_checksum = asset.current_checksum) -> + { t with selection = Some selected }, [] | _ -> let effects = release_file t @ release t @ if managed asset then [ demand selected ] else [] diff --git a/app/journal_media_runtime.ml b/app/journal_media_runtime.ml index fdb9c455..35ad6321 100644 --- a/app/journal_media_runtime.ml +++ b/app/journal_media_runtime.ml @@ -135,9 +135,14 @@ let notify t g = { token = c.consumer ; asset = c.asset ; file_type = - (match c.asset.source with - | Managed (Some v) -> v.file_type - | _ -> "") + (match P.presentation c.state with + | File path when Filename.extension path <> "" -> + let extension = Filename.extension path in + String.sub extension 1 (String.length extension - 1) + | _ -> + (match c.asset.source with + | Managed (Some v) -> v.file_type + | _ -> "")) ; presentation = P.presentation c.state }) (List.filter @@ -313,6 +318,25 @@ let release_import t (receipt : Logseq_db_worker.import_receipt) = receipt.preview ;; +let handoff_imports t g = + let completed, pending = + List.partition + (fun (receipt : Logseq_db_worker.import_receipt) -> + List.exists + (fun c -> + c.asset.uuid = receipt.asset.uuid + && c.asset.current_checksum = receipt.asset.current_checksum + && + match P.presentation c.state with + | P.File _ -> true + | _ -> false) + g.controllers) + g.local + in + g.local <- pending; + List.iter (release_import t) completed +;; + let clear t g = List.iter (release_import t) g.local; g.local <- []; @@ -1080,7 +1104,17 @@ let receive t ticket response = g.controllers <- controllers; g.cursor <- next_cursor; if not g.dependency_failed then g.error <- None; - List.iter (fun c -> Hashtbl.replace t.consumers c.consumer (g, c)) controllers + List.iter (fun c -> + Hashtbl.replace t.consumers c.consumer (g, c); + if g.visible && not c.shown + && List.exists (fun (receipt : Logseq_db_worker.import_receipt) -> + receipt.asset.uuid = c.asset.uuid) g.local + then match t.generation with + | None -> () + | Some graph_generation -> + c.shown <- true; + c.owners <- g.owners; + dispatch t g c (P.Show { graph_generation; consumer = c.consumer; asset = c.asset })) controllers | _ -> g.error <- Some "Unable to load attachments. Retry."); notify t g | _ -> ()) @@ -1095,6 +1129,7 @@ let receive t ticket response = (match List.find_opt (fun c -> c.consumer = consumer) g.controllers with | Some c -> dispatch t g c (Acquired (ticket, result)); + handoff_imports t g; notify t g | None -> release_stale t ticket result) | _ -> release_stale t ticket result)); @@ -1111,6 +1146,20 @@ let notice t scope = function notify t g; pump t | _ -> ()) + | Upload_status { operation; target; status = Cancelled; _ } -> + Option.iter + (fun g -> + let cancelled, pending = + List.partition + (fun (receipt : Logseq_db_worker.import_receipt) -> + receipt.operation = operation && receipt.scope = scope) + g.local + in + g.local <- pending; + List.iter (release_import t) cancelled; + notify t g; + pump t) + (Hashtbl.find_opt t.groups (G.Uuid.to_string target)) | Upload_status _ -> () | Asset_capacity_available -> Hashtbl.iter (fun _ (g, c) -> dispatch t g c P.Capacity_available) t.consumers; diff --git a/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md b/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md new file mode 100644 index 00000000..0d0b2f4b --- /dev/null +++ b/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md @@ -0,0 +1,51 @@ +# Capture Local Assets + +## Problem + +Capture drops a valid imported-file lease because its completion has no Detail route owner. The durable original is not published in the ordinary asset cache, and pending same-UUID media does not refresh its demand when metadata changes. A saved Capture can therefore show Waiting for upload despite an available original and later download its own upload. + +## Proposal + +Admit current Capture completions, resolve local files through the existing scoped durable upload intent, and publish verified originals into the ordinary asset cache through Core.Asset_requested and the sealed runner. Split private lookup and fetch tickets so Fetch always requires a remote version; use the exact strong spec diff approved on 2026-10-16. Refresh pending media demand when its version identity changes while preserving an already displayed file. No new persistent state, business phase, dependencies, Dune changes, or public worker callback changes. + +## Decision + +Use the approved strong ticket split and scoped local publication path. Keep durable upload intents as the sole persisted mapping, preserve successful displayed files, and retire temporary previews on handoff or matching cancellation. + +## Alternatives considered + +### Optional remote version on the shared transfer ticket + +Initially approved, then superseded by explicit approval of private lookup_ticket/fetch_ticket and lookup-source variants. A shared optional version admits a Fetch without a remote version and relies on runtime guards. + +### Retain the imported preview indefinitely + +This does not support cold start or group eviction and leaves lease cleanup tied to a transient UI group. Use the existing durable intent and ordinary cache instead. + +## Acceptance criteria + +- A saved Capture displays its original before PUT completes, with no asset GET. +- Pending local assets resolve offline, including after restart, in their exact account and graph scope. +- A missing local original never causes a GET without an explicit remote version; cancelled and stale completions cannot expose another scope's file. +- Publishing an original preserves active leases, survives cache reopen, and allows staging cleanup without redownloading the original. +- Pending same-UUID metadata changes refresh demand; an already displayed file remains stable. +- Public owner regressions, existing lifecycle suites, and isolated synthetic Simulator checks pass. No real upload or private graph is used. + +## Risks + +- Local and remote checksums may differ. Local lookup must use the current checksum and remote fetch must use the explicit remote version. +- Publication and release must preserve staging pins and fenced graph/account lifecycles; publication failure must not discard the source. +- Simulator host-restamped OCaml objects provide experimental runtime evidence, not clean iOS cross-compilation or ABI acceptance. + +## Consequences + +- The existing upload intent remains the durable identity authority. Its existing scope reconciliation also rebuilds a private in-memory asset/checksum index; lookup does not rescan every checkpoint for every visible attachment. +- Import attempts verified cache publication before returning. A pending lookup can use the retained original when cache capacity is exhausted. Successful terminal cleanup secures ordinary residency before retiring staging; failure preserves the source for recovery. +- A preview hands off only after the ordinary controller acquires a File. Cancelled notifications release an unpublished preview only for the matching scope and operation; failed or waiting uploads retain local display. +- Public owner regressions cover offline local readiness, missing local files, checksum changes, stale completions, scope mismatch, cold lookup, publication and lease lifetime, metadata refresh, preview handoff, and cancellation. Existing Application, upload, transfer, worker, cache, and sealed-interface checks pass. +- Isolated Simulator checks show both an old pending Capture and a newly saved Capture as final Timeline images before any asset PUT or GET, including after cold restart. The same installed binary later completed both encrypted PUTs and full metadata publication acknowledgements against a loopback synthetic service. Both intents reached Complete, staging was cleaned, and a subsequent cold start acquired the two ordinary cached originals with zero asset GETs. The service applied received transactions through public Database interfaces and computed acknowledgement checksums through public snapshot activation; no product completion events or persisted completion states were injected. This does not establish real cloud upload or full iOS cross-compilation/ABI acceptance. + +## Questions + +- May the two protected spec interfaces use the displayed strong ticket split and Publish_staged_asset request? Answered: explicitly approved by the user on 2026-10-16, replacing the optional-ticket proposal. +- May implementation use an isolated synthetic Simulator without touching private graphs or real network uploads? Answered: authorized; coordinate build and GUI windows with the other tasks. diff --git a/logseq_db_worker/lib/effect_runner/effect_runner.ml b/logseq_db_worker/lib/effect_runner/effect_runner.ml index 34a6b4cd..69b6f347 100644 --- a/logseq_db_worker/lib/effect_runner/effect_runner.ml +++ b/logseq_db_worker/lib/effect_runner/effect_runner.ml @@ -366,6 +366,7 @@ type t = ; mutable next_asset_operation : int64 ; upload_staging_lock : Eio.Mutex.t ; mutable staging_reconciled : Sync.graph_scope option + ; local_imports : (Graph.Uuid.t * string, Graph.Uuid.t) Hashtbl.t ; recovery_current : Core.upload_recovery_ticket -> bool ; upload_current : Upload.ticket -> bool ; asset_current : Logseq_sync_pure_reducer.Asset_transfer.ticket -> bool @@ -403,6 +404,7 @@ let create ~sw dependencies ~post ~recovery_current ~upload_current ~asset_curre ; next_asset_operation = 0L ; upload_staging_lock = Eio.Mutex.create () ; staging_reconciled = None + ; local_imports = Hashtbl.create 32 ; databases = Hashtbl.create 4 ; inspections = Hashtbl.create 4 ; waiters = Hashtbl.create 32 @@ -572,6 +574,7 @@ let reconcile_staging t (scope : Sync.graph_scope) ~current = else if t.staging_reconciled = Some scope then Ok () else ( + Hashtbl.clear t.local_imports; let keep = with_upload_store t (fun db -> let rec read after reversed = @@ -590,7 +593,11 @@ let reconcile_staging t (scope : Sync.graph_scope) ~current = let reversed = List.fold_left (fun files intent -> - intent.Logseq_db_types.Asset_upload_intent.staged_file :: files) + let module I = Logseq_db_types.Asset_upload_intent in + if intent.I.phase <> I.Cancelled then + Hashtbl.replace t.local_imports + (intent.asset, intent.version.checksum) intent.operation_id; + intent.staged_file :: files) reversed intents in @@ -1920,7 +1927,30 @@ let run_upload t (context : Sync.asset_context) instruction = | Upload.Cancel_operation _ -> () | Release_staging intent -> (match - asset_unit t ~scope:context.scope (Sync.Release_staged_file intent.staged_file) + let can_release = + if intent.Logseq_db_types.Asset_upload_intent.phase = Cancelled + then Ok () + else ( + match + asset_call + t + ~scope:context.scope + (Sync.Check_asset_cache (intent.asset, intent.version)) + with + | Ok (Sync.Asset_cached (Some handle)) -> + asset_unit t ~scope:context.scope (Sync.Release_asset_file handle) + | _ -> + asset_unit + t + ~scope:context.scope + (Sync.Publish_staged_asset + { asset = intent.asset + ; version = intent.version + ; file = intent.staged_file + })) + in + Result.bind can_release (fun () -> + asset_unit t ~scope:context.scope (Sync.Release_staged_file intent.staged_file)) with | Ok () -> () | Error message -> t.dependencies.publish (Core.Diagnostic message)) @@ -2041,13 +2071,74 @@ let transfer_failure = function | Asset_invalid_content message -> Invalid_content message ;; +let local_import t ~(scope : Sync.graph_scope) ~asset ~checksum ~current = + with_staging_lock t (fun () -> + let ( let* ) = Result.bind in + let* () = reconcile_staging t scope ~current in + match Hashtbl.find_opt t.local_imports (asset, checksum) with + | None -> Ok None + | Some operation -> + let* intent = + with_upload_store t (fun db -> + Logseq_db_storage.Asset_upload_store.read db ~operation) + in + Ok + (Option.bind intent (fun (i : Logseq_db_types.Asset_upload_intent.t) -> + if + current () + && (not t.stopped) + && i.origin = Uri.to_string scope.account.managed_sync_origin + && i.account = scope.account.user_id + && i.graph = scope.graph_id + && i.asset = asset + && i.version.checksum = checksum + && i.phase <> Logseq_db_types.Asset_upload_intent.Cancelled + then Some i + else None))) +;; + +let lookup_local_asset t ~scope ~asset ~checksum ~current call = + match local_import t ~scope ~asset ~checksum ~current with + | Error _ -> Error Sync.Asset_cancelled + | Ok None -> Ok (Sync.Asset_cached None) + | Ok (Some intent) -> + (match call (Sync.Check_asset_cache (asset, intent.version)) with + | Ok (Sync.Asset_cached (Some _) as value) -> Ok value + | Error _ as error -> error + | Ok _ -> + (match call (Sync.Retain_staged_file intent.staged_file) with + | Ok (Sync.Asset_retained (Some (lease, _))) -> + let published = + call + (Sync.Publish_staged_asset + { asset; version = intent.version; file = intent.staged_file }) + in + (match published with + | Error Sync.Asset_storage_full -> Ok (Sync.Asset_cached (Some lease)) + | _ -> + ignore (call (Sync.Release_asset_file lease)); + (match published with + | Ok Sync.Asset_unit -> + call (Sync.Check_asset_cache (asset, intent.version)) + | Error (Sync.Asset_missing_source | Asset_checksum_mismatch) -> + Ok (Sync.Asset_cached None) + | Error _ as error -> error + | Ok _ -> + Error (Sync.Asset_invalid_content "Unexpected publication completion"))) + | Ok _ -> Ok (Sync.Asset_cached None) + | Error _ as error -> error)) +;; + let run_asset t (context : Sync.asset_context) instruction = let module Transfer = Logseq_sync_pure_reducer.Asset_transfer in let perform ticket action success = - if - (not t.stopped) - && t.asset_current ticket - && not (Hashtbl.mem t.asset_operations ticket) + let scope = + match ticket with + | Transfer.Lookup ticket -> ticket.scope + | Download ticket -> ticket.scope + in + let valid () = (not t.stopped) && scope = context.scope && t.asset_current ticket in + if valid () && not (Hashtbl.mem t.asset_operations ticket) then ( Hashtbl.add t.asset_operations ticket ""; t.dependencies.runtime.fork ~sw:t.sw (fun () -> @@ -2055,7 +2146,7 @@ let run_asset t (context : Sync.asset_context) instruction = ~finally:(fun () -> Hashtbl.remove t.asset_operations ticket) (fun () -> let result = - if t.stopped || not (t.asset_current ticket) + if not (valid ()) then Error Sync.Asset_cancelled else asset_call @@ -2063,9 +2154,9 @@ let run_asset t (context : Sync.asset_context) instruction = Hashtbl.replace t.asset_operations ticket operation) t ~scope:context.scope - action + |> action in - if (not t.stopped) && t.asset_current ticket + if valid () then t.post (Core.Asset_completed @@ -2082,9 +2173,27 @@ let run_asset t (context : Sync.asset_context) instruction = in match instruction with | Transfer.Check_cache ticket -> + let work = Transfer.Lookup ticket in perform - ticket - (Sync.Check_asset_cache (ticket.asset, ticket.version)) + work + (fun call -> + let local checksum = + lookup_local_asset + t + ~scope:context.scope + ~asset:ticket.asset + ~checksum + ~current:(fun () -> t.asset_current work) + call + in + match ticket.source with + | Transfer.Local_checksum checksum -> local checksum + | Remote_version version -> call (Sync.Check_asset_cache (ticket.asset, version)) + | Local_and_remote { current_checksum; remote_version } -> + (match local current_checksum with + | Ok (Sync.Asset_cached None) -> + call (Sync.Check_asset_cache (ticket.asset, remote_version)) + | result -> result)) (fun result -> Transfer.Cache_checked ( ticket @@ -2093,12 +2202,14 @@ let run_asset t (context : Sync.asset_context) instruction = | _ -> Error (Transfer.Invalid_content "Unexpected cache completion")) )) | Fetch ticket -> perform - ticket - (Sync.Fetch_asset - { asset = ticket.asset - ; version = ticket.version - ; maximum_plaintext_bytes = 8 * 1024 * 1024 - }) + (Transfer.Download ticket) + (fun call -> + call + (Sync.Fetch_asset + { asset = ticket.asset + ; version = ticket.version + ; maximum_plaintext_bytes = 8 * 1024 * 1024 + })) (fun result -> Transfer.Downloaded ( ticket @@ -2346,8 +2457,27 @@ let prepare_import_unlocked result ;; -let prepare_import t ~context request ~current = - with_staging_lock t (fun () -> prepare_import_unlocked t ~context request ~current) +let prepare_import t ~(context : Sync.asset_context) request ~current = + with_staging_lock t (fun () -> + Result.map + (fun (intent : Logseq_db_types.Asset_upload_intent.t) -> + Hashtbl.replace + t.local_imports + (intent.asset, intent.version.checksum) + intent.operation_id; + if intent.phase <> Logseq_db_types.Asset_upload_intent.Cancelled && current () + then + ignore + (asset_unit + t + ~scope:context.Sync.scope + (Sync.Publish_staged_asset + { asset = intent.asset + ; version = intent.version + ; file = intent.staged_file + })); + intent) + (prepare_import_unlocked t ~context request ~current)) ;; let retain_imported_file t ~(scope : Sync.graph_scope) ~operation = diff --git a/logseq_db_worker/test/test_effect_runner.ml b/logseq_db_worker/test/test_effect_runner.ml index 38a6426d..cfbb4364 100644 --- a/logseq_db_worker/test/test_effect_runner.ml +++ b/logseq_db_worker/test/test_effect_runner.ml @@ -1020,6 +1020,7 @@ let test_upload_checkpoint_io () = let bridge_outputs = ref 0 in let io_runner = ref None in let fixture_output = ref None in + let transfer_ticket = ref None in let submit_sync = ref (fun _ -> failwith "sync interpreter not initialized") in let post = function | Core.Sync_event event -> @@ -1081,7 +1082,7 @@ let test_upload_checkpoint_io () = ~post ~recovery_current:(fun _ -> false) ~upload_current:(fun ticket -> !current && U.ticket_current state ticket) - ~asset_current:(fun _ -> false) + ~asset_current:(fun ticket -> !transfer_ticket = Some ticket) |> Result.get_ok in active_runner := Some runner; @@ -1191,6 +1192,46 @@ let test_upload_checkpoint_io () = "import is durable before success" true (durable = Some prepared && prepared.phase = I.Prepared); + (* Worker import owns publishing the staged original. This exercises its + public prepare operation and the sealed Core/runner bridge. *) + let normal_handle = + match fixture_asset (Sync.Check_asset_cache (prepared.asset, prepared.version)) with + | Sync.Asset_cached (Some handle) -> handle + | _ -> Alcotest.fail "imported original must be in normal cache before upload" + in + let normal_lease, normal_path = + match fixture_asset (Sync.Retain_asset_file normal_handle) with + | Sync.Asset_retained (Some file) -> file + | _ -> Alcotest.fail "normal imported original must be retainable" + in + Alcotest.(check string) "normal cache contains the original" "file" + (In_channel.with_open_bin normal_path In_channel.input_all); + ignore (fixture_asset (Sync.Release_asset_file normal_lease)); + ignore (fixture_asset (Sync.Release_asset_file normal_handle)); + let local_descriptor = Logseq_db_types.Asset_descriptor.create + ~uuid:prepared.asset ~source:(Managed None) + ~current_checksum:(Some prepared.version.checksum) ~size:None ~dimensions:None + |> Result.get_ok in + let module Transfer = Logseq_sync_pure_reducer.Asset_transfer in + let transfer = Transfer.create + (Transfer.config ~active:2 ~foreground_reserved:1 ~pending:3 ~retries:0 |> Result.get_ok) + ~scope ~online:false ~unlocked:false in + let _, instructions = Transfer.step transfer + (Replace { consumer = "local-original"; priority = Foreground; assets = [local_descriptor] }) in + let lookup = List.find_map (function Transfer.Check_cache ticket -> Some ticket | _ -> None) instructions |> Option.get in + transfer_ticket := Some (Transfer.Lookup lookup); + posted := []; + Runner.submit runner (Core.Run_asset (context, Transfer.Check_cache lookup)); + let local_handle = List.find_map (function + | Core.Asset_completed (actual, Transfer.Cache_checked (_, Ok (Some handle))) when actual = scope -> Some handle + | _ -> None) !posted |> Option.get in + ignore (fixture_asset (Sync.Release_asset_file local_handle)); + posted := []; + let before_wrong_scope = !bridge_requests in + Runner.submit runner (Core.Run_asset ({ context with scope = { scope with graph_id = uuid 90 } }, Transfer.Check_cache lookup)); + Alcotest.(check int) "mismatched transfer context performs no asset IO" before_wrong_scope !bridge_requests; + Alcotest.(check bool) "mismatched transfer context has no completion" true (!posted = []); + transfer_ticket := None; let lease, preview = match Runner.retain_imported_file runner ~scope ~operation:source.operation with | Some preview -> preview @@ -1318,10 +1359,18 @@ let test_upload_checkpoint_io () = ~post ~recovery_current:(fun _ -> false) ~upload_current:(fun _ -> false) - ~asset_current:(fun _ -> false) + ~asset_current:(fun ticket -> !transfer_ticket = Some ticket) |> Result.get_ok in active_runner := Some restarted; + transfer_ticket := Some (Transfer.Lookup lookup); + posted := []; + Runner.submit restarted (Core.Run_asset (context, Transfer.Check_cache lookup)); + let cold_handle = List.find_map (function + | Core.Asset_completed (actual, Transfer.Cache_checked (_, Ok (Some handle))) when actual = scope -> Some handle + | _ -> None) !posted |> Option.get in + ignore (fixture_asset (Sync.Release_asset_file cold_handle)); + transfer_ticket := None; let restored_import = Runner.prepare_import restarted ~context source ~current:(fun () -> true) |> Result.get_ok diff --git a/logseq_sync/lib/effect_runner/effect_runner.ml b/logseq_sync/lib/effect_runner/effect_runner.ml index a5e9b4cf..7986e534 100644 --- a/logseq_sync/lib/effect_runner/effect_runner.ml +++ b/logseq_sync/lib/effect_runner/effect_runner.ml @@ -1867,6 +1867,25 @@ let execute_asset ?except_id t (request : Core.asset_io_request) ~current = ~maximum_plaintext_bytes ~current |> map_unit) + | Publish_staged_asset { asset; version; file } -> + let ( let* ) = Result.bind in + cache (fun cache -> + match Asset_cache.retain_staged cache ~file with + | None -> Error Core.Asset_missing_source + | Some lease -> + Fun.protect + ~finally:(fun () -> Asset_cache.release cache lease) + (fun () -> + match Asset_cache.staged_path cache ~file with + | None -> Error Core.Asset_missing_source + | Some source_file -> + let* plaintext = read_upload_source + ~maximum_plaintext_bytes:t.dependencies.local_store.asset_maximum_file_bytes + source_file in + let* handle = Asset_cache.publish cache ~asset ~version ~current ~plaintext + |> Result.map_error asset_cache_failure in + Asset_cache.release cache handle; + Ok Core.Asset_unit)) | Retain_asset_file handle -> Ok (Core.Asset_retained (retain_asset_file t ~scope ~handle)) | Retain_staged_file file -> @@ -2062,6 +2081,7 @@ let cleanup_asset_action = function | Fetch_asset _ | Stage_asset_file _ | Put_asset_file _ + | Publish_staged_asset _ | Retain_asset_file _ | Retain_staged_file _ | Prune_asset_staging _ diff --git a/logseq_sync/lib/pure_reducer/asset_transfer.ml b/logseq_sync/lib/pure_reducer/asset_transfer.ml index d506e629..d679126d 100644 --- a/logseq_sync/lib/pure_reducer/asset_transfer.ml +++ b/logseq_sync/lib/pure_reducer/asset_transfer.ml @@ -29,16 +29,35 @@ type availability = ; retry_scheduled : bool } -type ticket = +type lookup_source = + | Local_checksum of string + | Remote_version of Asset.version + | Local_and_remote of + { current_checksum : string + ; remote_version : Asset.version + } + +type lookup_ticket = + { id : int + ; scope : Core.graph_scope + ; asset : Logseq_db_types.Graph_types.Uuid.t + ; source : lookup_source + } + +type fetch_ticket = { id : int ; scope : Core.graph_scope ; asset : Logseq_db_types.Graph_types.Uuid.t ; version : Asset.version } +type ticket = + | Lookup of lookup_ticket + | Download of fetch_ticket + type instruction = - | Check_cache of ticket - | Fetch of ticket + | Check_cache of lookup_ticket + | Fetch of fetch_ticket | Cancel of ticket | Release_handle of handle | Retry_after of @@ -61,8 +80,8 @@ type event = } | Release of string | Descriptor_changed of Asset.t - | Cache_checked of ticket * (handle option, failure) result - | Downloaded of ticket * (handle, failure) result + | Cache_checked of lookup_ticket * (handle option, failure) result + | Downloaded of fetch_ticket * (handle, failure) result | Retry of Logseq_db_types.Graph_types.Uuid.t | Retry_elapsed of int | Network_changed of bool @@ -82,8 +101,8 @@ type config = type phase = | Pending - | Checking of ticket - | Fetching of ticket + | Checking of lookup_ticket + | Fetching of fetch_ticket | Cached of handle | Rejected of failure * int option @@ -130,6 +149,21 @@ let create config ~scope ~online ~unlocked = } ;; +let lookup_source (asset : Asset.t) = + match asset.source, asset.current_checksum with + | Managed None, Some checksum -> Some (Local_checksum checksum) + | Managed (Some version), None -> Some (Remote_version version) + | Managed (Some remote_version), Some current_checksum -> + Some (Local_and_remote { current_checksum; remote_version }) + | Managed None, None | External _, _ -> None +;; + +let remote_version (asset : Asset.t) = + match asset.source with + | Managed version -> version + | External _ -> None +;; + let status t entry = match entry.phase with | Cached handle -> Ready handle @@ -139,17 +173,20 @@ let status t entry = | Rejected (failure, retry) -> Failed { failure; attempts = entry.attempts; retry_scheduled = Option.is_some retry } | Pending -> - (match entry.descriptor.source with - | Managed None -> Waiting_remote - | External _ -> Waiting_remote - | Managed (Some _) -> - if not entry.checked - then Queued - else if not t.online - then Waiting_network - else if not t.unlocked - then Waiting_unlock - else Queued) + if (not entry.checked) && Option.is_some (lookup_source entry.descriptor) + then Queued + else ( + match entry.descriptor.source with + | Managed None -> Waiting_remote + | External _ -> Waiting_remote + | Managed (Some _) -> + if not entry.checked + then Queued + else if not t.online + then Waiting_network + else if not t.unlocked + then Waiting_unlock + else Queued) ;; let availability t ~consumer = @@ -175,7 +212,8 @@ let pending_count t = let cleanup entry = match entry.phase with - | Checking ticket | Fetching ticket -> [ Cancel ticket ] + | Checking ticket -> [ Cancel (Lookup ticket) ] + | Fetching ticket -> [ Cancel (Download ticket) ] | Cached handle -> [ Release_handle handle ] | Pending | Rejected _ -> [] ;; @@ -201,9 +239,14 @@ let active entry = | _ -> false ;; -let fresh t asset version = +let fresh_lookup t asset source = + let id = t.serial + 1 in + { t with serial = id }, ({ id; scope = t.scope; asset; source } : lookup_ticket) +;; + +let fresh_fetch t asset version = let id = t.serial + 1 in - { t with serial = id }, { id; scope = t.scope; asset; version } + { t with serial = id }, ({ id; scope = t.scope; asset; version } : fetch_ticket) ;; let pump t = @@ -219,9 +262,9 @@ let pump t = |> List.filter (fun (_, e) -> e.phase = Pending && - match e.descriptor.source with - | Managed (Some _) -> (not e.checked) || (t.online && t.unlocked) - | Managed None | External _ -> false) + if e.checked + then Option.is_some (remote_version e.descriptor) && t.online && t.unlocked + else Option.is_some (lookup_source e.descriptor)) |> List.stable_sort (fun (_, a) (_, b) -> Bool.compare (foreground b) (foreground a)) in let t, _, _, effects = @@ -233,20 +276,28 @@ let pump t = || ((not fg) && background >= t.config.active - t.config.foreground_reserved) then t, count, background, effects else ( - match entry.descriptor.source with - | Managed (Some version) -> - let t, ticket = fresh t uuid version in - let phase, instruction = - if entry.checked - then Fetching ticket, Fetch ticket - else Checking ticket, Check_cache ticket - in - let entry = { entry with phase } in - ( { t with entries = Assets.add uuid entry t.entries } + let work = + if entry.checked + then + Option.map + (fun version -> + let t, ticket = fresh_fetch t uuid version in + t, Fetching ticket, Fetch ticket) + (remote_version entry.descriptor) + else + Option.map + (fun source -> + let t, ticket = fresh_lookup t uuid source in + t, Checking ticket, Check_cache ticket) + (lookup_source entry.descriptor) + in + match work with + | None -> t, count, background, effects + | Some (t, phase, instruction) -> + ( { t with entries = Assets.add uuid { entry with phase } t.entries } , count + 1 , (background + if fg then 0 else 1) - , instruction :: effects ) - | Managed None | External _ -> t, count, background, effects)) + , instruction :: effects ))) (t, count, background, []) candidates in @@ -255,7 +306,8 @@ let pump t = let same_version a b = match a.Asset.source, b.Asset.source with - | Managed a, Managed b -> a = b + | Managed a_source, Managed b_source -> + a_source = b_source && a.current_checksum = b.current_checksum | External a, External b -> a = b | _ -> false ;; @@ -345,10 +397,18 @@ let replace t consumer priority descriptors = else candidate, effects ;; -let valid_ticket t (ticket : ticket) expected = - ticket.scope = t.scope && ticket = expected +let ticket_asset = function + | Lookup t -> t.asset + | Download t -> t.asset +;; + +let ticket_scope = function + | Lookup t -> t.scope + | Download t -> t.scope ;; +let valid_ticket t ticket expected = ticket_scope ticket = t.scope && ticket = expected + let stale_handle t handle = if Assets.exists (fun _ entry -> entry.phase = Cached handle) t.entries then [] @@ -371,13 +431,14 @@ let fail t entry failure = else t, { entry with attempts; phase = Rejected (failure, None) }, [] ;; -let completed t ticket result cache = +let completed t ticket result = + let asset = ticket_asset ticket in let matching = - match Assets.find_opt ticket.asset t.entries with + match Assets.find_opt asset t.entries with | Some ({ phase = Checking expected; _ } as entry) - when cache && valid_ticket t ticket expected -> Some entry + when valid_ticket t ticket (Lookup expected) -> Some entry | Some ({ phase = Fetching expected; _ } as entry) - when (not cache) && valid_ticket t ticket expected -> Some entry + when valid_ticket t ticket (Download expected) -> Some entry | _ -> None in match matching with @@ -393,7 +454,7 @@ let completed t ticket result cache = | Ok None -> t, { entry with phase = Pending; checked = true }, [] | Error failure -> fail t entry failure in - { t with entries = Assets.add ticket.asset entry t.entries }, effects + { t with entries = Assets.add asset entry t.entries }, effects ;; let resume_entry entry = @@ -407,9 +468,9 @@ let raw_step t = function if t.stopped then t, [] else replace t consumer priority assets | Release consumer -> release t consumer | Descriptor_changed descriptor -> update_descriptor t descriptor - | Cache_checked (ticket, result) -> completed t ticket result true + | Cache_checked (ticket, result) -> completed t (Lookup ticket) result | Downloaded (ticket, result) -> - completed t ticket (Result.map Option.some result) false + completed t (Download ticket) (Result.map Option.some result) | Retry uuid -> (match Assets.find_opt uuid t.entries with | Some ({ phase = Rejected _; _ } as entry) -> @@ -440,7 +501,7 @@ let raw_step t = function let entry, effects = match entry.phase with | Fetching ticket when not online -> - { entry with phase = Pending }, Cancel ticket :: effects + { entry with phase = Pending }, Cancel (Download ticket) :: effects | _ -> entry, effects in Assets.add uuid entry entries, effects) @@ -455,7 +516,7 @@ let raw_step t = function let entry, effects = match entry.phase with | Fetching ticket when not unlocked -> - { entry with phase = Pending }, Cancel ticket :: effects + { entry with phase = Pending }, Cancel (Download ticket) :: effects | Rejected (Locked, _) when unlocked -> resume_entry entry, effects | _ -> entry, effects in @@ -505,11 +566,10 @@ let step t event = next, effects @ scheduled @ notifications t next @ capacity ;; -let ticket_current t (ticket : ticket) = - (not t.stopped) - && - match Assets.find_opt ticket.asset t.entries with - | Some { phase = Checking expected; _ } | Some { phase = Fetching expected; _ } -> - ticket = expected - | _ -> false +let ticket_current t ticket = + (not t.stopped) && ticket_scope ticket = t.scope + && match Assets.find_opt (ticket_asset ticket) t.entries with + | Some { phase = Checking expected; _ } -> ticket = Lookup expected + | Some { phase = Fetching expected; _ } -> ticket = Download expected + | _ -> false ;; diff --git a/logseq_sync/lib/pure_reducer/core.ml b/logseq_sync/lib/pure_reducer/core.ml index 1c75c46f..a7eb9c4b 100644 --- a/logseq_sync/lib/pure_reducer/core.ml +++ b/logseq_sync/lib/pure_reducer/core.ml @@ -334,6 +334,11 @@ type asset_action = ; file : string ; maximum_plaintext_bytes : int } + | Publish_staged_asset of + { asset : Logseq_db_types.Graph_types.Uuid.t + ; version : Logseq_db_types.Asset_descriptor.version + ; file : string + } | Retain_asset_file of string | Retain_staged_file of string | Release_asset_file of string @@ -3051,6 +3056,7 @@ let asset_cleanup_action = function | Fetch_asset _ | Stage_asset_file _ | Put_asset_file _ + | Publish_staged_asset _ | Retain_asset_file _ | Retain_staged_file _ | Prune_asset_staging _ @@ -3088,6 +3094,7 @@ let asset_result_matches (request : asset_io_request) = function | Fetch_asset _, Asset_downloaded _ | Stage_asset_file _, Asset_staged _ | Put_asset_file _, Asset_unit + | Publish_staged_asset _, Asset_unit | Retain_asset_file _, Asset_retained _ | Retain_staged_file _, Asset_retained _ | Release_asset_file _, Asset_unit diff --git a/logseq_sync/spec/pure_reducer/asset_transfer.mli b/logseq_sync/spec/pure_reducer/asset_transfer.mli index 5fc7408b..2ea81e3b 100644 --- a/logseq_sync/spec/pure_reducer/asset_transfer.mli +++ b/logseq_sync/spec/pure_reducer/asset_transfer.mli @@ -29,16 +29,35 @@ type availability = ; retry_scheduled : bool } -type ticket = private +type lookup_source = + | Local_checksum of string + | Remote_version of Asset.version + | Local_and_remote of + { current_checksum : string + ; remote_version : Asset.version + } + +type lookup_ticket = private + { id : int + ; scope : Core.graph_scope + ; asset : Logseq_db_types.Graph_types.Uuid.t + ; source : lookup_source + } + +type fetch_ticket = private { id : int ; scope : Core.graph_scope ; asset : Logseq_db_types.Graph_types.Uuid.t ; version : Asset.version } +type ticket = + | Lookup of lookup_ticket + | Download of fetch_ticket + type instruction = - | Check_cache of ticket - | Fetch of ticket + | Check_cache of lookup_ticket + | Fetch of fetch_ticket | Cancel of ticket | Release_handle of handle | Retry_after of @@ -61,8 +80,8 @@ type event = } | Release of string | Descriptor_changed of Asset.t - | Cache_checked of ticket * (handle option, failure) result - | Downloaded of ticket * (handle, failure) result + | Cache_checked of lookup_ticket * (handle option, failure) result + | Downloaded of fetch_ticket * (handle, failure) result | Retry of Logseq_db_types.Graph_types.Uuid.t | Retry_elapsed of int | Network_changed of bool diff --git a/logseq_sync/spec/pure_reducer/core.mli b/logseq_sync/spec/pure_reducer/core.mli index 1697a074..4c5c353c 100644 --- a/logseq_sync/spec/pure_reducer/core.mli +++ b/logseq_sync/spec/pure_reducer/core.mli @@ -239,6 +239,11 @@ type asset_action = ; file : string ; maximum_plaintext_bytes : int } + | Publish_staged_asset of + { asset : Logseq_db_types.Graph_types.Uuid.t + ; version : Logseq_db_types.Asset_descriptor.version + ; file : string + } | Retain_asset_file of string | Retain_staged_file of string | Release_asset_file of string diff --git a/logseq_sync/test/runner_contract.ml b/logseq_sync/test/runner_contract.ml index 0361c7ed..f4f99ac8 100644 --- a/logseq_sync/test/runner_contract.ml +++ b/logseq_sync/test/runner_contract.ml @@ -2190,3 +2190,83 @@ let scenarios = scenarios @ [Alcotest.test_case let scenarios = scenarios @ [Alcotest.test_case "asset retirement restore conflict is retryable" `Quick test_asset_retirement_restore_conflict] + +(* Cache publication owns verification, atomic residency, and lease lifetime; + the worker regression separately owns when an import requests publication. *) +let test_cache_publish_staged_original () = + with_local_asset_cache (fun support request -> + let source = Filename.concat support "picker.bin" in + Out_channel.with_open_bin source (fun out -> output_string out "file"); + let file, checksum, _ = cache_stage request ~file_type:"pdf" source in + let version = + Logseq_db_types.Asset_descriptor.version ~checksum ~file_type:"pdf" |> Result.get_ok + in + let staged_lease, staged_path = + request (Core.Retain_staged_file file) |> retained_value + in + cache_unit + (request (Core.Publish_staged_asset { asset = graph_id (); version; file })); + let handle = + match request (Core.Check_asset_cache (graph_id (), version)) with + | Ok (Core.Asset_cached (Some handle)) -> handle + | _ -> fail "published original missing from ordinary cache" + in + let lease, path = request (Core.Retain_asset_file handle) |> retained_value in + cache_bool + "ordinary cache keeps document extension" + true + (Filename.check_suffix path ".pdf"); + Alcotest.(check string) + "published original bytes" + "file" + (In_channel.with_open_bin path In_channel.input_all); + cache_unit (request (Core.Release_staged_file file)); + cache_bool + "publication preserves native staging lease" + true + (Sys.file_exists staged_path); + cache_unit (request (Core.Release_asset_file staged_lease)); + cache_bool "last staging lease permits cleanup" false (Sys.file_exists staged_path); + cache_bool "normal renderer survives staging cleanup" true (Sys.file_exists path); + cache_unit (request (Core.Release_asset_file lease)); + cache_unit (request (Core.Release_asset_file handle)); + cache_unit (request Core.Close_asset_scope); + let restored = + match request (Core.Check_asset_cache (graph_id (), version)) with + | Ok (Core.Asset_cached (Some handle)) -> handle + | _ -> fail "published original must survive cache reopen" + in + cache_unit (request (Core.Release_asset_file restored)); + let file, _, _ = cache_stage request source in + let wrong = + Logseq_db_types.Asset_descriptor.version + ~checksum:(String.make 64 'a') + ~file_type:"bin" + |> Result.get_ok + in + cache_bool + "wrong checksum cannot publish" + true + (request (Core.Publish_staged_asset { asset = graph_id (); version = wrong; file }) + = Error Core.Asset_checksum_mismatch); + let held, held_path = request (Core.Retain_staged_file file) |> retained_value in + cache_bool "failed publication preserves source" true (Sys.file_exists held_path); + cache_bool + "foreign filename is rejected" + true + (request + (Core.Publish_staged_asset + { asset = graph_id (); version; file = "../picker.bin" }) + = Error Core.Asset_missing_source); + cache_unit (request (Core.Release_asset_file held)); + cache_unit (request (Core.Release_staged_file file))) +;; + +let scenarios = + scenarios + @ [ Alcotest.test_case + "publish staged original into cache" + `Quick + test_cache_publish_staged_original + ] +;; diff --git a/logseq_sync/test/test_asset_transfer.ml b/logseq_sync/test/test_asset_transfer.ml index db7422ba..8951dad4 100644 --- a/logseq_sync/test/test_asset_transfer.ml +++ b/logseq_sync/test/test_asset_transfer.ml @@ -23,14 +23,14 @@ let scope : C.graph_scope = } ;; -let asset ?(checksum = String.make 64 'a') ?(remote = true) n = +let asset ?(checksum = String.make 64 'a') ?(remote = true) ?current_checksum n = get (A.create ~uuid:(uuid n) ~source: (Managed (if remote then Some (get (A.version ~checksum ~file_type:"png")) else None)) - ~current_checksum:None + ~current_checksum ~size:None ~dimensions:None) ;; @@ -250,7 +250,7 @@ let lock_and_network () = let t, e = T.step t (Cache_checked (only (checks e), Ok None)) in let old = only (fetches e) in let t, e = T.step t (Unlock_changed false) in - Alcotest.(check bool) "lock cancels IO" true (List.mem (T.Cancel old) e); + Alcotest.(check bool) "lock cancels IO" true (List.mem (T.Cancel (T.Download old)) e); let t, e = T.step t (Downloaded (old, Ok "stale")) in Alcotest.(check bool) "lock fences bytes" true (release_present e); let t, _ = T.step t (Network_changed false) in @@ -301,12 +301,12 @@ let duplicate_retry_while_loading () = let local = only (checks effects) in let t, effects = T.step t (Retry (uuid 2)) in check "retry while checking has no effects" 0 (List.length effects); - Alcotest.(check bool) "local lookup stays current" true (T.ticket_current t local); + Alcotest.(check bool) "local lookup stays current" true (T.ticket_current t (T.Lookup local)); let t, effects = T.step t (Cache_checked (local, Ok None)) in let download = only (fetches effects) in let t, effects = T.step t (Retry (uuid 2)) in check "retry while fetching has no effects" 0 (List.length effects); - Alcotest.(check bool) "download stays current" true (T.ticket_current t download) + Alcotest.(check bool) "download stays current" true (T.ticket_current t (T.Download download)) ;; let ready_does_not_refresh () = @@ -369,12 +369,12 @@ let manual_retry_fences_old_work () = Alcotest.(check bool) "fresh local ticket survives stale completion" true - (T.ticket_current t local); + (T.ticket_current t (T.Lookup local)); let t, effects = T.step t (Release "visible") in Alcotest.(check bool) "release cancels local lookup" true - (List.mem (T.Cancel local) effects); + (List.mem (T.Cancel (T.Lookup local)) effects); let t, effects = T.step t (Cache_checked (local, Ok (Some "stale"))) in Alcotest.(check bool) "released completion cleans resource" @@ -400,6 +400,41 @@ let manual_retry_fences_old_work () = check "old scope cannot create demand" 0 (T.pending_count fresh) ;; +let local_before_remote () = + let a = asset ~remote:false ~current_checksum:(String.make 64 'a') 2 in + let t, effects = replace (initial ~online:false ()) "visible" Foreground [ a ] in + check "pending local original is checked offline" 1 (List.length (checks effects)); + let t, effects = T.step t (Cache_checked (only (checks effects), Ok (Some "original"))) in + Alcotest.(check bool) "local original is Ready before upload" true (ready t "visible"); + check "local original never downloads" 0 (List.length (fetches effects)) +;; + +let local_miss_waits_for_publication () = + let a = asset ~remote:false ~current_checksum:(String.make 64 'a') 2 in + let t, effects = replace (initial ()) "visible" Foreground [ a ] in + check "pending original gets one lookup" 1 (List.length (checks effects)); + let t, effects = T.step t (Cache_checked (only (checks effects), Ok None)) in + check "missing local original does not GET" 0 (List.length (fetches effects)); + Alcotest.(check bool) "missing local waits for remote version" true + (T.availability t ~consumer:"visible" = [ uuid 2, T.Waiting_remote ]); + let t, effects = T.step t (Network_changed true) in + check "network cannot manufacture remote version" 0 (List.length (checks effects) + List.length (fetches effects)); + let _, effects = T.step t (Descriptor_changed (asset 2)) in + check "publication starts fresh lookup" 1 (List.length (checks effects)) +;; + +let local_checksum_fences_lookup () = + let local c = asset ~remote:false ~current_checksum:(String.make 64 c) 2 in + let t, effects = replace (initial ()) "visible" Foreground [ local 'a' ] in + check "local checksum starts lookup" 1 (List.length (checks effects)); + let old = only (checks effects) in + let t, effects = T.step t (Descriptor_changed (local 'b')) in + check "new local checksum starts fresh lookup" 1 (List.length (checks effects)); + let t, effects = T.step t (Cache_checked (old, Ok (Some "stale"))) in + Alcotest.(check bool) "old local completion releases lease" true (release_present effects); + Alcotest.(check bool) "old local completion cannot become Ready" false (ready t "visible") +;; + let () = Alcotest.run "asset transfer" @@ -414,6 +449,9 @@ let () = ; "foreground reserve", foreground ; "backpressure", backpressure ; "offline cache", offline + ; "local before upload offline", local_before_remote + ; "local miss waits for publication", local_miss_waits_for_publication + ; "local checksum fences lookup", local_checksum_fences_lookup ; "bounded 404 retry", retry_limit ; "duplicate completion", duplicate ; "pending foreground capacity", foreground_queue_capacity diff --git a/test/application_view_test.ml b/test/application_view_test.ml index 2d6f6526..188de6fb 100644 --- a/test/application_view_test.ml +++ b/test/application_view_test.ml @@ -462,6 +462,7 @@ let run_favorites_native_visibility let import_probe = Atomic.make false in let import_probed = Atomic.make false in let staged_paths = ref [] in + let durable_paths = Atomic.make [] in let detail_captures = Atomic.make [] in let child_attempts = Atomic.make 0 in let child_release = @@ -733,19 +734,14 @@ let run_favorites_native_visibility else Ok (Service.Asset_imported (Error "Fixture first attachment failed"))) else let module A = Logseq_db_types.Asset_descriptor in + let durable_path = source.source_file ^ ".durable" in + let bytes = "synthetic local attachment" in + Out_channel.with_open_bin durable_path (fun out -> output_string out bytes); + Atomic.set durable_paths (durable_path :: Atomic.get durable_paths); let asset = - A.create - ~uuid:source.asset - ~source: - (Managed - (Some - (A.version - ~checksum:(String.make 64 'a') - ~file_type:source.file_type - |> Result.get_ok))) - ~current_checksum:None - ~size:None - ~dimensions:None + A.create ~uuid:source.asset ~source:(Managed None) + ~current_checksum:(Some (Logseq_sync_effect_runner.Asset_codec.checksum bytes)) + ~size:(Some (Int64.of_int (String.length bytes))) ~dimensions:None |> Result.get_ok in Atomic.incr imports_finished; @@ -758,7 +754,7 @@ let run_favorites_native_visibility ; target = source.target ; asset ; file_type = source.file_type - ; preview = None + ; preview = Some ("synthetic-capture-local-lease", durable_path) })) | Import_asset _ -> Error "unexpected import" | Graph_request request -> @@ -1339,7 +1335,7 @@ let run_favorites_native_visibility Atomic.set acquire_release true; ignore (hooks.dispose ()); Option.iter Logseq_db_worker_lui.Journal_worker_runtime.stop !client; - List.iter (fun path -> if Sys.file_exists path then Sys.remove path) !staged_paths; + List.iter (fun path -> if Sys.file_exists path then Sys.remove path) (!staged_paths @ Atomic.get durable_paths); check_fixture_worker_idle ()) (fun () -> hooks.init 2 2 startup |> consume; @@ -3367,6 +3363,15 @@ let run_favorites_native_visibility match !staged_paths with | [failed; successful] -> Sys.file_exists failed && not (Sys.file_exists successful) | _ -> false); + settle_worker (); + Alcotest.(check bool) "durable original remains present" true + (List.for_all Sys.file_exists (Atomic.get durable_paths)); + (* Application owns async owner admission. A pure navigation/media + test cannot execute this completion gate; returning a valid receipt + through the public Worker service reproduces it without injecting + an incorrect current flag. No upload or network operation exists. *) + Alcotest.(check int) "Capture retains local lease before any upload" 0 + (Atomic.get released_files); Alcotest.(check (list string)) "batch preserves picker order" [ "First attachment"; "Second attachment" ] diff --git a/test/journal_media_runtime_test.ml b/test/journal_media_runtime_test.ml index 0ce3e4f2..b5d9e18a 100644 --- a/test/journal_media_runtime_test.ml +++ b/test/journal_media_runtime_test.ml @@ -1117,3 +1117,200 @@ let () = "real membership removal releases the retained File lease" ) ] ;; + +(* Runtime owns the imported receipt lease and the handoff to a normal media + controller; neither the pure media nor transfer reducer owns this lease. *) +let () = + let sent = Queue.create () + and views = Hashtbl.create 1 in + let runtime = + R.create + ~send:(fun ticket request -> + Queue.add (ticket, request) sent; + true) + ~changed:(fun root view -> Hashtbl.replace views root view) + in + R.reset runtime ~graph_generation:(Some 1); + let root = G.Uuid.to_string (uuid 1) in + let scope : S.asset_scope = + { account = + { managed_sync_origin = Uri.of_string "https://fixture.invalid" + ; user_id = "fixture" + ; account_generation = 1 + ; presentation_generation = 1 + ; lifecycle_generation = 1L + } + ; graph_id = uuid 3 + ; graph_generation = 1 + } + in + let asset = + A.create + ~uuid:(uuid 2) + ~source:(Managed None) + ~current_checksum:(Some (String.make 64 'a')) + ~size:None + ~dimensions:None + |> Result.get_ok + in + let receipt : Logseq_db_worker.import_receipt = + { operation = uuid 8 + ; graph_generation = 1 + ; scope + ; target = uuid 1 + ; asset + ; file_type = "png" + ; preview = Some ("staged-lease", "/staged/original.png") + } + in + R.imported runtime ~current:true receipt; + R.root_visible runtime ~root true; + let ticket, request = Queue.take sent in + let request_id = + match request with + | S.Graph_request q -> q.request_id + | _ -> failwith "query" + in + R.receive + runtime + (Option.get ticket) + (Graph_response + (P.V2_response + { api_version = 2 + ; request_id + ; outcome = + V2_assets_outcome + { generation = "g" + ; projection_revision = "p" + ; items = [ asset ] + ; next_cursor = None + } + })); + let requests = Queue.to_seq sent |> List.of_seq in + let consumer = + List.find_map + (function + | _, S.Asset_command { command = S.Replace_asset_demand { consumer; _ }; _ } -> + Some consumer + | _ -> None) + requests + in + check (Option.is_some consumer) "imported metadata must start normal local demand"; + check + ((List.hd (Hashtbl.find views root).items).presentation + = Journal_media.File "/staged/original.png") + "handoff keeps preview while acquisition is pending"; + Queue.clear sent; + let consumer = Option.get consumer in + R.notice + runtime + scope + (Asset_availability + { consumer; asset = asset.uuid; availability = Ready "normal-cache" }); + let ticket, request = Queue.take sent in + check + (match request with + | S.Acquire_asset_file _ -> true + | _ -> false) + "normal file is acquired"; + R.receive + runtime + (Option.get ticket) + (Asset_file (Some ("normal-lease", "/cache/original.png"))); + let items = (Hashtbl.find views root).items in + check + (List.length items = 1 + && (List.hd items).presentation = Journal_media.File "/cache/original.png") + "normal file replaces staged preview without duplicate items"; + check + ((List.hd items).file_type = "png") + "pending local image keeps its type after handoff"; + check + (List.exists + (function + | _, S.Release_asset_file { handle = "staged-lease"; _ } -> true + | _ -> false) + (Queue.to_seq sent |> List.of_seq)) + "normal acquisition releases staged preview"; + Queue.clear sent; + R.reset runtime ~graph_generation:None; + check + (not + (List.exists + (function + | _, S.Release_asset_file { handle = "staged-lease"; _ } -> true + | _ -> false) + (Queue.to_seq sent |> List.of_seq))) + "handoff releases staged preview only once" +;; + +let () = + let sent = Queue.create () + and views = Hashtbl.create 1 in + let runtime = + R.create + ~send:(fun ticket request -> + Queue.add (ticket, request) sent; + true) + ~changed:(fun root view -> Hashtbl.replace views root view) + in + R.reset runtime ~graph_generation:(Some 1); + let scope : S.asset_scope = + { account = + { managed_sync_origin = Uri.of_string "https://fixture.invalid" + ; user_id = "fixture" + ; account_generation = 1 + ; presentation_generation = 1 + ; lifecycle_generation = 1L + } + ; graph_id = uuid 3 + ; graph_generation = 1 + } + in + let asset = + A.create + ~uuid:(uuid 2) + ~source:(Managed None) + ~current_checksum:(Some (String.make 64 'a')) + ~size:None + ~dimensions:None + |> Result.get_ok + in + let receipt : Logseq_db_worker.import_receipt = + { operation = uuid 8 + ; graph_generation = 1 + ; scope + ; target = uuid 1 + ; asset + ; file_type = "png" + ; preview = Some ("cancelled-preview", "/staged/original.png") + } + in + let root = G.Uuid.to_string receipt.target in + let notice = + S.Upload_status + { operation = receipt.operation + ; asset = asset.uuid + ; target = receipt.target + ; title = "fixture" + ; status = Cancelled + } + in + R.imported runtime ~current:true receipt; + R.notice runtime { scope with graph_id = uuid 90 } notice; + check + (List.length (Hashtbl.find views root).items = 1 && Queue.is_empty sent) + "foreign scope cannot cancel imported preview"; + R.notice runtime scope notice; + check ((Hashtbl.find views root).items = []) "cancelled unpublished preview disappears"; + check + (List.exists + (function + | _, S.Release_asset_file { handle = "cancelled-preview"; _ } -> true + | _ -> false) + (Queue.to_seq sent |> List.of_seq)) + "cancellation releases its staged preview"; + Queue.clear sent; + R.notice runtime scope notice; + check (Queue.is_empty sent) "duplicate cancellation releases no second lease" +;; diff --git a/test/journal_media_test.ml b/test/journal_media_test.ml index 8d7368db..a72e4233 100644 --- a/test/journal_media_test.ml +++ b/test/journal_media_test.ml @@ -112,6 +112,19 @@ let () = incr failed; Printf.printf "FAIL %s: %s\n%!" name (Printexc.to_string exn) in + run "pending same-UUID metadata refreshes demand" (fun () -> + let local checksum = A.create ~uuid:(uuid 1) ~source:(Managed None) + ~current_checksum:(Some (String.make 64 checksum)) ~size:None ~dimensions:None + |> Result.get_ok in + let pending, _ = P.step P.empty (show (local 'a')) in + let _, unchanged = P.step pending (show (local 'a')) in + check (unchanged = []) "unchanged pending descriptor is inert"; + let _, effects = P.step pending (show (asset 'a')) in + check (List.exists (function P.Demand _ -> true | _ -> false) effects) + "published version refreshes pending demand"; + let _, effects = P.step pending (show (local 'b')) in + check (List.exists (function P.Demand _ -> true | _ -> false) effects) + "changed current checksum refreshes pending demand"); run "typed local readiness shows Loading until acquisition" (fun () -> let waiting, _ = P.step P.empty (show (asset 'a')) in let pending, effects = ready waiting "local" in From 9d535a01de32eb247b67f87873db568ff22fe56a Mon Sep 17 00:00:00 2001 From: rcmerci Date: Fri, 16 Oct 2026 15:59:38 +0800 Subject: [PATCH 2/4] Keep imported previews open during cache handoff --- app/journal_media_runtime.ml | 24 +++++++-- .../bugfix/2026-10-16-capture-local-assets.md | 3 +- test/journal_media_runtime_test.ml | 53 ++++++++++++++++++- 3 files changed, 73 insertions(+), 7 deletions(-) diff --git a/app/journal_media_runtime.ml b/app/journal_media_runtime.ml index 35ad6321..a0a5c238 100644 --- a/app/journal_media_runtime.ml +++ b/app/journal_media_runtime.ml @@ -322,7 +322,11 @@ let handoff_imports t g = let completed, pending = List.partition (fun (receipt : Logseq_db_worker.import_receipt) -> - List.exists + let token = "import:" ^ G.Uuid.to_string receipt.operation in + not + (Hashtbl.to_seq_values t.previews + |> Seq.exists (fun (_, root, asset) -> root = g.root && asset = token)) + && List.exists (fun c -> c.asset.uuid = receipt.asset.uuid && c.asset.current_checksum = receipt.asset.current_checksum @@ -497,8 +501,11 @@ let retain_owners t owners = Owners.iter (fun owner -> if not (Owners.mem owner retained) then root_visible ~owner t ~root false) - g.owners) - t.groups + g.owners; + handoff_imports t g; + notify t g) + t.groups; + pump t ;; let asset_visible ?(owner = "default") t ~root ~asset visible = @@ -539,7 +546,16 @@ let preview_visible t ~owner ~slot ~root ~asset visible = if (not visible) || current <> Some selected then ( Hashtbl.remove t.previews preview; - Option.iter (fun (_, root, _) -> root_visible ~owner:preview t ~root false) current); + Option.iter + (fun (_, root, _) -> + root_visible ~owner:preview t ~root false; + Option.iter + (fun g -> + handoff_imports t g; + notify t g) + (Hashtbl.find_opt t.groups root)) + current; + pump t); if visible then ( match Hashtbl.find_opt t.groups root with diff --git a/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md b/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md index 0d0b2f4b..88da4c76 100644 --- a/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md +++ b/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md @@ -29,6 +29,7 @@ This does not support cold start or group eviction and leaves lease cleanup tied - A missing local original never causes a GET without an explicit remote version; cancelled and stale completions cannot expose another scope's file. - Publishing an original preserves active leases, survives cache reopen, and allows staging cleanup without redownloading the original. - Pending same-UUID metadata changes refresh demand; an already displayed file remains stable. +- An open imported preview keeps its token, path and lease until its final preview owner closes; background cache handoff must not dismiss it. - Public owner regressions, existing lifecycle suites, and isolated synthetic Simulator checks pass. No real upload or private graph is used. ## Risks @@ -41,7 +42,7 @@ This does not support cold start or group eviction and leaves lease cleanup tied - The existing upload intent remains the durable identity authority. Its existing scope reconciliation also rebuilds a private in-memory asset/checksum index; lookup does not rescan every checkpoint for every visible attachment. - Import attempts verified cache publication before returning. A pending lookup can use the retained original when cache capacity is exhausted. Successful terminal cleanup secures ordinary residency before retiring staging; failure preserves the source for recovery. -- A preview hands off only after the ordinary controller acquires a File. Cancelled notifications release an unpublished preview only for the matching scope and operation; failed or waiting uploads retain local display. +- A preview hands off only after the ordinary controller acquires a File and no open preview still owns the imported token. Explicit close and owner cleanup finish the handoff; multiple previews retain the original until the final one closes. Cancelled notifications release an unpublished preview only for the matching scope and operation; failed or waiting uploads retain local display. - Public owner regressions cover offline local readiness, missing local files, checksum changes, stale completions, scope mismatch, cold lookup, publication and lease lifetime, metadata refresh, preview handoff, and cancellation. Existing Application, upload, transfer, worker, cache, and sealed-interface checks pass. - Isolated Simulator checks show both an old pending Capture and a newly saved Capture as final Timeline images before any asset PUT or GET, including after cold restart. The same installed binary later completed both encrypted PUTs and full metadata publication acknowledgements against a loopback synthetic service. Both intents reached Complete, staging was cleaned, and a subsequent cold start acquired the two ordinary cached originals with zero asset GETs. The service applied received transactions through public Database interfaces and computed acknowledgement checksums through public snapshot activation; no product completion events or persisted completion states were injected. This does not establish real cloud upload or full iOS cross-compilation/ABI acceptance. diff --git a/test/journal_media_runtime_test.ml b/test/journal_media_runtime_test.ml index b5d9e18a..6897185f 100644 --- a/test/journal_media_runtime_test.ml +++ b/test/journal_media_runtime_test.ml @@ -1119,8 +1119,8 @@ let () = ;; (* Runtime owns the imported receipt lease and the handoff to a normal media - controller; neither the pure media nor transfer reducer owns this lease. *) -let () = + controller and open-preview references; neither pure reducer owns these. *) +let test_import_handoff preview = let sent = Queue.create () and views = Hashtbl.create 1 in let runtime = @@ -1200,6 +1200,16 @@ let () = ((List.hd (Hashtbl.find views root).items).presentation = Journal_media.File "/staged/original.png") "handoff keeps preview while acquisition is pending"; + let preview_token = (List.hd (Hashtbl.find views root).items).token in + let show owner = + R.preview_visible runtime ~owner ~slot:"selected-import" ~root ~asset:preview_token true + and close owner = + R.preview_visible runtime ~owner ~slot:"selected-import" ~root ~asset:preview_token false + in + (match preview with + | `None -> () + | `Close | `Prune | `Multiple -> show "row"); + if preview = `Multiple then show "detail"; Queue.clear sent; let consumer = Option.get consumer in R.notice @@ -1217,6 +1227,35 @@ let () = runtime (Option.get ticket) (Asset_file (Some ("normal-lease", "/cache/original.png"))); + let release_count () = + Queue.to_seq sent + |> Seq.filter (function + | _, S.Release_asset_file { handle = "staged-lease"; _ } -> true + | _ -> false) + |> Seq.length + in + let preview_is_retained () = + let item = List.hd (Hashtbl.find views root).items in + check + (item.token = preview_token + && item.presentation = Journal_media.File "/staged/original.png" + && release_count () = 0) + "an open import preview keeps its token, path and lease during cache handoff" + in + (match preview with + | `None -> () + | `Close -> + preview_is_retained (); + close "row"; + close "row" + | `Prune -> + preview_is_retained (); + R.retain_owners runtime [ "default" ] + | `Multiple -> + preview_is_retained (); + close "row"; + preview_is_retained (); + close "detail"); let items = (Hashtbl.find views root).items in check (List.length items = 1 @@ -1232,6 +1271,7 @@ let () = | _ -> false) (Queue.to_seq sent |> List.of_seq)) "normal acquisition releases staged preview"; + check (release_count () = 1) "handoff or final preview close releases its lease once"; Queue.clear sent; R.reset runtime ~graph_generation:None; check @@ -1244,6 +1284,15 @@ let () = "handoff releases staged preview only once" ;; +let () = + run_cases + [ "import handoff without preview", (fun () -> test_import_handoff `None) + ; "open import preview survives handoff", (fun () -> test_import_handoff `Close) + ; "import preview owner cleanup", (fun () -> test_import_handoff `Prune) + ; "import handoff waits for final preview", (fun () -> test_import_handoff `Multiple) + ] +;; + let () = let sent = Queue.create () and views = Hashtbl.create 1 in From b0ba455f1cb41f60af0299da3a2b4be04eb27b07 Mon Sep 17 00:00:00 2001 From: rcmerci Date: Fri, 16 Oct 2026 16:08:16 +0800 Subject: [PATCH 3/4] Release hidden import preview leases after final owner closes --- app/journal_media_runtime.ml | 21 +++++++------ .../bugfix/2026-10-16-capture-local-assets.md | 3 +- test/journal_media_runtime_test.ml | 30 +++++++++++++------ 3 files changed, 35 insertions(+), 19 deletions(-) diff --git a/app/journal_media_runtime.ml b/app/journal_media_runtime.ml index a0a5c238..7a3f9297 100644 --- a/app/journal_media_runtime.ml +++ b/app/journal_media_runtime.ml @@ -326,15 +326,18 @@ let handoff_imports t g = not (Hashtbl.to_seq_values t.previews |> Seq.exists (fun (_, root, asset) -> root = g.root && asset = token)) - && List.exists - (fun c -> - c.asset.uuid = receipt.asset.uuid - && c.asset.current_checksum = receipt.asset.current_checksum - && - match P.presentation c.state with - | P.File _ -> true - | _ -> false) - g.controllers) + (* Hidden controllers already released their File. Without a view + owner, the durable intent/cache owns the bytes, not this UI lease. *) + && ((not g.visible) + || List.exists + (fun c -> + c.asset.uuid = receipt.asset.uuid + && c.asset.current_checksum = receipt.asset.current_checksum + && + match P.presentation c.state with + | P.File _ -> true + | _ -> false) + g.controllers)) g.local in g.local <- pending; diff --git a/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md b/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md index 88da4c76..092ab9a4 100644 --- a/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md +++ b/docs/agent-guide/implemented/bugfix/2026-10-16-capture-local-assets.md @@ -30,6 +30,7 @@ This does not support cold start or group eviction and leaves lease cleanup tied - Publishing an original preserves active leases, survives cache reopen, and allows staging cleanup without redownloading the original. - Pending same-UUID metadata changes refresh demand; an already displayed file remains stable. - An open imported preview keeps its token, path and lease until its final preview owner closes; background cache handoff must not dismiss it. +- Closing or pruning the final preview owner releases its UI lease even after the source row hides; the durable intent/cache continues to own the bytes. - Public owner regressions, existing lifecycle suites, and isolated synthetic Simulator checks pass. No real upload or private graph is used. ## Risks @@ -42,7 +43,7 @@ This does not support cold start or group eviction and leaves lease cleanup tied - The existing upload intent remains the durable identity authority. Its existing scope reconciliation also rebuilds a private in-memory asset/checksum index; lookup does not rescan every checkpoint for every visible attachment. - Import attempts verified cache publication before returning. A pending lookup can use the retained original when cache capacity is exhausted. Successful terminal cleanup secures ordinary residency before retiring staging; failure preserves the source for recovery. -- A preview hands off only after the ordinary controller acquires a File and no open preview still owns the imported token. Explicit close and owner cleanup finish the handoff; multiple previews retain the original until the final one closes. Cancelled notifications release an unpublished preview only for the matching scope and operation; failed or waiting uploads retain local display. +- A visible preview hands off only after the ordinary controller acquires a File and no open preview still owns the imported token. Explicit close and owner cleanup finish the handoff; multiple previews retain the original until the final one closes. If the source row is hidden, the final preview owner releases the UI lease without requiring an already hidden controller to retain a File; the durable intent/cache continues to own the bytes. Cancelled notifications release an unpublished preview only for the matching scope and operation; failed or waiting uploads retain local display. - Public owner regressions cover offline local readiness, missing local files, checksum changes, stale completions, scope mismatch, cold lookup, publication and lease lifetime, metadata refresh, preview handoff, and cancellation. Existing Application, upload, transfer, worker, cache, and sealed-interface checks pass. - Isolated Simulator checks show both an old pending Capture and a newly saved Capture as final Timeline images before any asset PUT or GET, including after cold restart. The same installed binary later completed both encrypted PUTs and full metadata publication acknowledgements against a loopback synthetic service. Both intents reached Complete, staging was cleaned, and a subsequent cold start acquired the two ordinary cached originals with zero asset GETs. The service applied received transactions through public Database interfaces and computed acknowledgement checksums through public snapshot activation; no product completion events or persisted completion states were injected. This does not establish real cloud upload or full iOS cross-compilation/ABI acceptance. diff --git a/test/journal_media_runtime_test.ml b/test/journal_media_runtime_test.ml index 6897185f..64ac4742 100644 --- a/test/journal_media_runtime_test.ml +++ b/test/journal_media_runtime_test.ml @@ -1208,7 +1208,7 @@ let test_import_handoff preview = in (match preview with | `None -> () - | `Close | `Prune | `Multiple -> show "row"); + | `Close | `Prune | `Multiple | `Hidden_close | `Hidden_prune -> show "row"); if preview = `Multiple then show "detail"; Queue.clear sent; let consumer = Option.get consumer in @@ -1255,15 +1255,25 @@ let test_import_handoff preview = preview_is_retained (); close "row"; preview_is_retained (); - close "detail"); + close "detail" + | `Hidden_close | `Hidden_prune -> + R.root_visible runtime ~root false; + preview_is_retained (); + if preview = `Hidden_close then close "row" else R.retain_owners runtime []; + check + (release_count () = 1 + && (List.hd (Hashtbl.find views root).items).token <> preview_token) + "the final preview owner releases its import even after the row is hidden"); let items = (Hashtbl.find views root).items in - check - (List.length items = 1 - && (List.hd items).presentation = Journal_media.File "/cache/original.png") - "normal file replaces staged preview without duplicate items"; - check - ((List.hd items).file_type = "png") - "pending local image keeps its type after handoff"; + if preview <> `Hidden_close && preview <> `Hidden_prune + then ( + check + (List.length items = 1 + && (List.hd items).presentation = Journal_media.File "/cache/original.png") + "normal file replaces staged preview without duplicate items"; + check + ((List.hd items).file_type = "png") + "pending local image keeps its type after handoff"); check (List.exists (function @@ -1290,6 +1300,8 @@ let () = ; "open import preview survives handoff", (fun () -> test_import_handoff `Close) ; "import preview owner cleanup", (fun () -> test_import_handoff `Prune) ; "import handoff waits for final preview", (fun () -> test_import_handoff `Multiple) + ; "hidden import preview close", (fun () -> test_import_handoff `Hidden_close) + ; "hidden import preview owner cleanup", (fun () -> test_import_handoff `Hidden_prune) ] ;; From ac42df90f7de684b894ce9c6a23220a4797ed181 Mon Sep 17 00:00:00 2001 From: rcmerci Date: Fri, 16 Oct 2026 16:29:16 +0800 Subject: [PATCH 4/4] Adapt worker lifecycle test to tagged lookup tickets --- logseq_db_worker/test/test_pure_reducer.ml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/logseq_db_worker/test/test_pure_reducer.ml b/logseq_db_worker/test/test_pure_reducer.ml index ca1586ca..71feb55e 100644 --- a/logseq_db_worker/test/test_pure_reducer.ml +++ b/logseq_db_worker/test/test_pure_reducer.ml @@ -399,7 +399,7 @@ let test_asset_demand_lifecycle () = Alcotest.(check bool) "live ticket" true - (Core.asset_ticket_current requested.next ticket); + (Core.asset_ticket_current requested.next (Transfer.Lookup ticket)); let completed = Core.step requested.next