Preserve concrete error causes across Journal boundaries

app/impl/worker_coordination/worker_effect_authority.mli

@@ -439,7 +439,7 @@ val maximum_mutation_recovery_bytes : int
 
 type interruption =
   | Cancelled
-  | Transport_lost
+  | Transport_lost of error
   | Worker_stopped
 
 type 'a reply =

app/impl/worker_coordination/worker_effect_runner.ml

@@ -212,6 +212,15 @@ let submit t = function
   | Schedule_drain -> t.post Core.Drain
 ;;
 
+let retirement_result operation decode = function
+  | Mailbox.Completed response -> decode response
+  | Failed message -> Error (Worker.Service_failed (operation ^ ": " ^ message))
+  | Cancelled ->
+    Error (Worker.Service_failed (operation ^ ": cancelled before retirement acknowledgement"))
+  | Shutdown ->
+    Error (Worker.Service_failed (operation ^ ": worker stopped before retirement acknowledgement"))
+;;
+
 let on_event t event =
   let runtime, worker =
     match event with
@@ -288,7 +297,10 @@ let on_event t event =
                      Worker.Answered (Error e))
                 | Failed message ->
                   (match request with
-                   | Worker.Import _ -> Worker.Interrupted Worker.Transport_lost
+                   | Worker.Import _ ->
+                     let cause = Worker.Service_failed ("Import: " ^ message) in
+                     error := Some cause;
+                     Worker.Interrupted (Worker.Transport_lost cause)
                    | _ -> Worker.Answered (Error (Worker.Service_failed message)))
                 | Cancelled -> Worker.Interrupted Worker.Cancelled
                 | Shutdown -> Worker.Interrupted Worker.Worker_stopped
@@ -301,19 +313,14 @@ let on_event t event =
             | Unobserve (issuer, session, id) ->
               let module I = (val issuer : Worker_effect_authority.ISSUER) in
               let result =
-                match outcome with
-                | Completed response ->
-                  I.observation_registration_released session ~observation_id:id response
-                | _ ->
-                  Error
-                    (Worker.Service_failed
-                       "Observation release did not acknowledge physical retirement")
+                retirement_result "Observation release"
+                  (I.observation_registration_released session ~observation_id:id) outcome
               in
               t.post (Core.Observation_release_completed (session, id, result))
             |Finalize_observation(issuer,session,id)->
               let module I=(val issuer:Worker_effect_authority.ISSUER) in
-              let result=match outcome with Completed response->I.observation_registration_finalized session ~observation_id:id response
-                |_->Error(Worker.Service_failed "Observation finalization was not acknowledged") in
+              let result=retirement_result "Observation finalization"
+                (I.observation_registration_finalized session ~observation_id:id) outcome in
               t.post(Core.Observation_finalize_completed(session,id,result))
             |Release_demand(issuer,ticket,registration)->
               let module I=(val issuer:Worker_effect_authority.ISSUER) in
@@ -357,14 +364,7 @@ let on_event t event =
                 |Ok receipt->Worker.Safe_to_proceed receipt|Error error->Worker.Handoff_failed(change,error)))
             | Unregister issuer ->
               let module I = (val issuer : Worker_effect_authority.ISSUER) in
-              let result =
-                match outcome with
-                | Completed response -> I.binding_released response
-                | _ ->
-                  Error
-                    (Worker.Service_failed
-                       "Unregister did not acknowledge resource retirement")
-              in
+              let result = retirement_result "Binding unregister" I.binding_released outcome in
               t.post (Core.Binding_release_completed (I.binding, result)));
            Consumed)
       | Push { payload = Service.Graph_pushes _ as payload; push_sequence; _ } ->
@@ -410,7 +410,7 @@ let on_event t event =
             | None->()
             | Some resident->(match I.asset_notice resident.session data with
               | Ok notice->t.post(Core.Asset_received(notice,push_sequence))
-              | Error _->()))(Service.asset_notices payload))
+              | Error cause->error := Some cause))(Service.asset_notices payload))
           (Core.projection view))t.latest;
         Broadcast
       | Push _ -> Broadcast

app/impl/worker_coordination/worker_pure_api.ml

@@ -121,7 +121,7 @@ type admission_error =
 
 type interruption =
   | Cancelled
-  | Transport_lost
+  | Transport_lost of error
   | Worker_stopped
 
 type 'a reply =

app/impl/worker_coordination/worker_pure_coordinator.ml

@@ -2557,6 +2557,13 @@ let plan_host_action_attempt (state:t) receipt =
     host_consume_early=None;host_consume_waiting=false},Consume_handoff(receipt,attempt))
 ;;
 
+let cleanup_admission_error operation = function
+  | Worker.Full -> None
+  | Not_ready -> Some (Worker.Service_failed (operation ^ ": worker not ready"))
+  | Stopping -> Some (Worker.Service_failed (operation ^ ": worker stopping"))
+  | Unsupported_request reason -> Some (Worker.Unsupported_operation (operation ^ ": " ^ reason))
+;;
+
 let step_base state event =
   match event with
   | Import_source_proof_returned proof ->
@@ -3002,11 +3009,7 @@ let step_base state event =
      | Some (Observing value)
        when Worker.equal_session session value.obs_session && value.obs_release_inflight && not value.obs_release_done
        ->
-       let error =
-         match admission with
-         | Worker.Full -> None
-         | _ -> Some (Worker.Service_failed "Observation cleanup not admitted")
-       in
+       let error = cleanup_admission_error "Observation cleanup" admission in
        changed
          { state with
            facts =
@@ -3058,7 +3061,7 @@ let step_base state event =
     (match Observation_ids.find_opt id state.facts.observations with
      |Some(Observing value) when value.obs_release_done && value.obs_release_inflight
        && Worker.equal_session session value.obs_session->
-       let error=if admission=Worker.Full then None else Some(Worker.Service_failed "Observation finalization not admitted") in
+       let error=cleanup_admission_error "Observation finalization" admission in
        let observations=Observation_ids.add id (Observing{value with obs_release_inflight=false;
          obs_cleanup_waiting=admission=Worker.Full;obs_cleanup_error=error})state.facts.observations in
        changed {state with facts={state.facts with observations}} []

app/journal_asset_import.ml

@@ -165,11 +165,16 @@ let extension =
 
 (* Staged picks live under the host temp directory as [journal-import-*];
    delete only files matching that contract so a user path is never removed. *)
-let discard_staged_file (staged : staged) =
+let discard_staged_file ?(on_failure=fun message->
+    try prerr_endline ("journal: operation=discardStagedFile " ^ message) with _->()) (staged : staged) =
   if String.starts_with ~prefix:"journal-import-" (Filename.basename staged.source_file)
   then (
-    try Sys.remove staged.source_file with
-    | _ -> ())
+    try Unix.unlink staged.source_file with
+    | Unix.Unix_error(Unix.ENOENT,_,_)->()
+    | error->
+      let module D=Logseq_db_types.Failure_diagnostic in
+      let detail=D.of_source ~operation:"assetRelease" ~reason:"cleanupFailed" (Printexc.to_string error) in
+      on_failure(D.encode detail))
 ;;
 
 let is_dismissal payload =

app/journal_asset_import.mli

@@ -50,8 +50,10 @@ type event =
 val decode_event : string -> (event, string) result
 
 (** Best-effort removal of a staged temp copy created for this pick; a no-op
-    for paths outside the staged temp-file naming contract. *)
-val discard_staged_file : staged -> unit
+    for paths outside the staged temp-file naming contract or already absent files.
+    Other failures reach [on_failure] as a bounded diagnostic carrier; the default
+    writes the same safe cause to the local diagnostic stream. *)
+val discard_staged_file : ?on_failure:(string -> unit) -> staged -> unit
 
 val is_dismissal : string -> bool
 

app/journal_bridge.ml

@@ -85,7 +85,77 @@ let platform_failure payload = (hooks ()).platform_failure payload
 let dispose () = (hooks ()).dispose ()
 let root_node () = (hooks ()).root_node ()
 
+type initialization_stage = Decode_startup | Start_worker | Initialize
+
+(* Stable bounded codes shared with JournalRuntime; zero is an explicit unknown.
+   No strings from producers or exceptions are retained by the exception. *)
+exception Initialization_failure of initialization_stage * int
+
+let initialization_cause = function
+  | "startup payload is truncated" -> 1
+  | "startup payload exceeds 1 MiB" -> 2
+  | "invalid startup magic" -> 3
+  | "startup configuration length is invalid" -> 4
+  | "startup payload has trailing or missing bytes" -> 5
+  | "startup configuration must be valid UTF-8 JSON" -> 6
+  | "invalid startup configuration: application_support_directory must be a bounded canonical absolute path" -> 7
+  | "invalid startup configuration: response_budget_bytes is outside the protocol bounds" -> 8
+  | "invalid startup configuration: default_page_size is outside the protocol bounds" -> 9
+  | "invalid startup configuration: sync base URL must be one HTTPS origin without credentials or fragments" -> 10
+  | "worker.start.failure" -> 12
+  | "worker.start.invalidArgument" -> 13
+  | "worker.start.systemError" -> 14
+  | "worker.start.permissionDenied" -> 15
+  | "worker.start.missingFile" -> 16
+  | "worker.start.noSpace" -> 17
+  | "worker.start.busy" -> 18
+  | "worker.start.io" -> 19
+  | "worker.start.unixError" -> 20
+  | "worker.start.outOfMemory" -> 21
+  | "worker.start.stackOverflow" -> 22
+  | "worker.start.endOfFile" -> 23
+  | "worker.start.matchFailure" -> 24
+  | "worker.start.assertFailure" -> 25
+  | "worker.start.notFound" -> 26
+  | "application.mount.failed" -> 27
+  | "Worker Domain subsystem is stopped" | "Worker Domain has stopped" -> 28
+  | "Worker Domain subsystem is terminal" -> 29
+  | "Worker Domain session is already attached" -> 30
+  | "Worker Domain did not start" -> 31
+  | _ -> 0
+;;
+
+let initialization_failure stage source =
+  Initialization_failure (stage, initialization_cause source)
+;;
+
+let initialization_exception = function
+  | Initialization_failure _ as error -> error
+  | error ->
+    let cause = match error with
+      | Failure message when message = "application.mount.failed" -> 27
+      | Failure _ -> 12
+      | Invalid_argument _ -> 13
+      | Sys_error _ -> 14
+      | Unix.Unix_error ((Unix.EACCES | Unix.EPERM), _, _) -> 15
+      | Unix.Unix_error (Unix.ENOENT, _, _) -> 16
+      | Unix.Unix_error (Unix.ENOSPC, _, _) -> 17
+      | Unix.Unix_error ((Unix.EBUSY | Unix.EAGAIN), _, _) -> 18
+      | Unix.Unix_error (Unix.EIO, _, _) -> 19
+      | Unix.Unix_error _ -> 20
+      | Out_of_memory -> 21
+      | Stack_overflow -> 22
+      | End_of_file -> 23
+      | Match_failure _ -> 24
+      | Assert_failure _ -> 25
+      | Not_found -> 26
+      | _ -> 0
+    in Initialization_failure (Initialize, cause)
+;;
+
 let classify_exception = function
+  | Initialization_failure (stage, cause) ->
+    (match stage with Decode_startup -> 500 | Start_worker -> 600 | Initialize -> 700) + cause
   | Failure message when message = "application.decodeStartup.failed" -> 5
   | Failure message when message = "application.startWorker.failed" -> 6
   | Failure message when message = "application.mount.failed" -> 7

app/journal_bridge.mli

@@ -61,3 +61,12 @@ val platform_failure_message : string -> string
 (** Recognized native diagnostic family and finite subcause. Unrecognized
     suffixes use a fixed unknown reason; arbitrary payloads are never echoed. *)
 val native_diagnostic : string -> (string * string) option
+
+
+(** Startup failures retain a finite operation and cause without exception
+    arguments, paths, configuration values or credentials. *)
+type initialization_stage = Decode_startup | Start_worker | Initialize
+val initialization_failure : initialization_stage -> string -> exn
+(** Preserve an existing typed startup failure, or classify an exception by its
+    constructor and a bounded Unix error category. *)
+val initialization_exception : exn -> exn

app/journal_detail.ml

@@ -1042,7 +1042,9 @@ module Pure_observation = struct
     | Not_admitted Stopping -> Some (Worker.Service_failed "Worker is stopping.")
     | Not_admitted (Unsupported_request message) -> Some (Worker.Unsupported_operation message)
     | Answered (Error error) -> Some error
-    | Interrupted _ -> Some Worker.Revoked_session
+    | Interrupted (Transport_lost error) -> Some error
+    | Interrupted Cancelled -> Some(Worker.Service_failed "Observation request was cancelled.")
+    | Interrupted Worker_stopped -> Some(Worker.Service_failed "Worker stopped during observation.")
     | Answered (Ok _) -> None
   ;;
 

app/journal_graph_favorites_plan.ml

@@ -27,6 +27,7 @@ type event =
   | Favorites_loaded of Journal_graph_request.favorites_request * Journal_worker_coordination.Worker_api.favorites
   | Favorites_failed of Journal_graph_request.favorites_request * bool * string
   | Favorites_invalidated
+  | Diagnostic of { operation : string; request_id : Graph.Uuid.t; error : Error.t }
   | Rejected of rejection
 type output =
   { requests : Graph.Uuid.t Rrbvec.t
@@ -509,7 +510,9 @@ let fail t operation ~request_id ~code ~message ~error =
   let rejected () = responses [ Rejected (Worker_rejection
     { operation = operation_name operation; request_id; error }) ] in
   match operation with
-  | List_favorites request -> responses [Favorites_failed (request, code = Error.Stale_read_cursor, message)]
+  | List_favorites request ->
+    responses [Favorites_failed (request, code = Error.Stale_read_cursor, message);
+      Diagnostic {operation=operation_name operation;request_id;error}]
   | Favorite_dependency {epoch;uuid;_} ->
     if epoch <> t.favorite_epoch then empty else complete_favorite_dependency t uuid (rejected ())
   | Favorite_memberships {epoch;page;_} ->

app/journal_graph_favorites_plan.mli

@@ -11,6 +11,7 @@ type event =
   | Favorites_loaded of Journal_graph_request.favorites_request * Journal_worker_coordination.Worker_api.favorites
   | Favorites_failed of Journal_graph_request.favorites_request * bool * string
   | Favorites_invalidated
+  | Diagnostic of { operation : string; request_id : Graph.Uuid.t; error : Error.t }
   | Rejected of rejection
 type output =
   { requests : Graph.Uuid.t Rrbvec.t

app/journal_graph_feed_plan.ml

@@ -262,8 +262,8 @@ let continue_root_batch t ~ports owner page cursor limit =
     then cursor_failure batch "The retained root page changed. Refresh Timeline."
     else (
       match ports.validate_session cursor.session with
-      | Error _ ->
-        cursor_failure batch "The retained root execution was revoked. Refresh Timeline."
+      | Error error ->
+        cursor_failure batch (Journal_worker_coordination_host.error_message error)
       | Ok () ->
         (match
            Worker.validate_request
@@ -277,8 +277,8 @@ let continue_root_batch t ~ports owner page cursor limit =
                     ; after = cursor.after
                     } ))
          with
-         | Error _ ->
-           cursor_failure batch "The retained root query no longer matches its budget."
+         | Error error ->
+           cursor_failure batch (Journal_worker_coordination_host.error_message error)
          | Ok () when not (Rrbvec.is_empty cursor.remaining) ->
            { empty with cached = Some batch }
          | Ok () when cursor.after <> None -> requests [ plan_root_batch ~ports batch ]
@@ -815,8 +815,8 @@ let resume_cached t ~ports batch =
     then superseded_root_batch t ~ports batch
     else (
       match ports.validate_session cursor.Projection.session with
-      | Error _ ->
-        cursor_failure batch "The retained root execution was revoked. Refresh Timeline."
+      | Error error ->
+        cursor_failure batch (Journal_worker_coordination_host.error_message error)
       | Ok () ->
         receive_root_tree
           t

app/journal_graph_mutation_confirmation.ml

@@ -301,7 +301,7 @@ module Pure_insertion = struct
     then Error "The atomic Capture children do not match the admitted payload."
     else Result.bind(Graph.Uuid.of_string command.mutation_id)(fun mutation_id->
       match Worker.journal_day day with
-      |Error _->Error "Capture requires its admitted day and atomic Append payload."
+      |Error error->Error(Journal_worker_coordination_host.error_message error)
       |Ok journal_day->Result.map(fun(t,action)->{t with capture=Some command},action)
         (start_value ~command:(capture_command command) ~page:None ~resolved_status ~append_day:(Some day)
           (Append{mutation_id;day=journal_day;root})))