diff --git a/docs/agent-guide/implemented/simplification/2026-10-17-remove-discarded-commit-callbacks.md b/docs/agent-guide/implemented/simplification/2026-10-17-remove-discarded-commit-callbacks.md new file mode 100644 index 00000000..cc8defcf --- /dev/null +++ b/docs/agent-guide/implemented/simplification/2026-10-17-remove-discarded-commit-callbacks.md @@ -0,0 +1,68 @@ +# Remove Discarded Commit Callbacks + +## Problem + +The three commit paths in `logseq_overlay_db/lib/database.ml` collect active +subscription callbacks under the database mutex, return that list through every +success/error branch, and immediately discard it with `ignore callbacks`. +The list is neither invoked nor retained. This repeats a subscription scan and +obscures the real notification owner. + +The reviewed baseline is main `fc27cfd62e7fede81185c7ec0bc8807fca797b52`. +`commit_local`, `apply_outbox_transition`, and `commit_authoritative_candidate` +all call `serialize_commit`. That function reserves dispatcher capacity before +the mutation, enqueues the committed change and awaits queue delivery. The +dispatcher selects active subscriptions and publishes to their streams; +`activate_subscription` owns callback execution and its lifecycle fences. + +## Proposal + +Remove only the discarded callback collections and the unused component of each +private return tuple. Keep the existing `(change, outcome)` results, persistence, +locks, dispatcher reservations, stream delivery and callback lifecycle unchanged. +No public interface, protected spec, Dune file, dependency or platform bridge +changes are required. The user authorized implementing both audited +simplifications, testing and opening a PR on 2026-10-17 +(Sentinel_b66872b73d108191964dce936fc35406). + +## Decision + +Adopt the proposal as authorized. The implementation is confined to the private +overlay implementation; the existing behavioral contracts remain unchanged. + +## Alternatives considered + +### Keep a callback snapshot at commit time + +A snapshot could support a different notification contract, but no current +consumer receives this list. Wiring it into delivery would change the existing +dispatcher and unlisten semantics. The existing stream owner remains authoritative. + +## Acceptance criteria + +- All three commit paths return the same success/error outcomes and change events. +- No-change/transport-only commits remain silent; concurrent commits publish in + revision order and reserve capacity before committing. +- Callback exceptions remain isolated. Self-unlisten, external unlisten and close + retain their current in-flight callback behavior. +- Existing `test_overlay_changes` and `test_overlay_concurrency` pass before and + after the deletion; final full build and tests pass. +- No new behavior is introduced, so existing owner tests are the regression + baseline; do not add tests that assert source spelling or unused tuple shapes. + +## Risks + +- Mechanical tuple edits must not change a branch result or the real dispatcher. + These are private implementation tuples, not a public contract cutover. + +## Consequences + +Commit paths no longer allocate discarded subscriber lists or carry an unused +private tuple component. Dispatcher capacity, notification streams and callback +lifecycle remain the sole delivery mechanism. No interface or persistence +format changes are introduced. + +## Questions + +None. The implementation and PR are explicitly authorized; no new product or +interface decision is needed. diff --git a/docs/agent-guide/implemented/simplification/2026-10-17-trim-replan-only-snapshot-work.md b/docs/agent-guide/implemented/simplification/2026-10-17-trim-replan-only-snapshot-work.md new file mode 100644 index 00000000..6702fa80 --- /dev/null +++ b/docs/agent-guide/implemented/simplification/2026-10-17-trim-replan-only-snapshot-work.md @@ -0,0 +1,74 @@ +# Trim Replan Only Snapshot Work + +## Problem + +`replan_queued_ordinary` in `logseq_overlay_db/lib/database.ml` reverses its +accumulated outbox for every queued ordinary mutation and builds a children +effect index whose per-parent lists append each record. Neither value is read +by this temporary snapshot's consumers, adding avoidable cumulative work. + +At main `fc27cfd62e7fede81185c7ec0bc8807fca797b52`, both authoritative commit and +stale-rejection settlement enter this private function. The temporary snapshot +only reaches `logical_snapshot_satisfies_mutation`, `invalid_replan_dependency` +and `ordinary_effect_footprint`. Their transitive block, page, asset-descriptor, +asset-version and tombstone readers consume the authoritative root and block/page +effect maps. They do not enter journal enumeration, child enumeration or public +snapshot dispatch. The normal snapshot outbox and children indexes have real +consumers elsewhere and remain required. + +## Proposal + +Remove this function's children-index construction and threaded accumulator. +Leave `outbox` and `children_effects` uncaptured (`None`) in this private +temporary snapshot, so an unexpected future structural read fails explicitly. +Preserve the actual final outbox accumulator, stable sequence order, block/page +effect buckets, active dependency shadows, tombstones, asset checks, normalized +transactions and result classifications. Do not redesign the shared snapshot or +modify its public interface, protected spec, Dune files or dependencies. + +The user authorized implementing both audited simplifications, testing and +opening a PR on 2026-10-17 +(Sentinel_b66872b73d108191964dce936fc35406). + +## Decision + +Adopt the proposal as authorized. The implementation is confined to the private +overlay implementation; the existing behavioral contracts remain unchanged. + +## Alternatives considered + +### Generalize a separate replan snapshot API + +A new representation would prevent future structural reads but require broad +reader/interface changes. The current snapshot remains private and the complete +local call chain is small enough to verify; a targeted deletion removes work +without moving complexity into another API. + +## Acceptance criteria + +- Existing ordinary rebase, no-change, transitive dependency blocking and stale + settlement tests pass before and after the deletion. +- Asset-reference replan, local asset descriptors, tombstones and chronological + block/page effect replay retain their current behavior. +- General journal, children and page-tree reads retain their real indexes. +- Existing overlay sync/asset/read/mutation cases and final full build/tests pass. +- No public interfaces or new persistence/scheduling mechanisms are introduced. + Add behavior coverage only if the existing cases leave a demonstrated gap. + +## Risks + +- A future call to a structural reader from this temporary path would need its + required index. Current call-chain review must include nested asset/tombstone + reads; the general snapshot producers must not be changed. + +## Consequences + +Ordinary replan no longer builds a children-effect map or repeatedly reverses +its retained prefix for an unread temporary snapshot field. Real block/page +effect buckets and the final retained record order remain authoritative. General +journal and structure snapshot producers retain their required roots. + +## Questions + +None. The implementation and PR are explicitly authorized; no new product or +interface decision is needed. diff --git a/logseq_overlay_db/lib/database.ml b/logseq_overlay_db/lib/database.ml index 0a855733..15e99f5b 100644 --- a/logseq_overlay_db/lib/database.ml +++ b/logseq_overlay_db/lib/database.ml @@ -5641,29 +5641,29 @@ let serialize_commit database operation = let commit_local database ~expected mutation = serialize_commit database (fun () -> - let callbacks, change, outcome = + let change, outcome = Eio.Mutex.use_rw ~protect:true database.lock (fun () -> if database.closed - then [], None, Error Types.Local_database_closed + then None, Error Types.Local_database_closed else ( match Overlay_planner.validate mutation with - | Error message -> [], None, Error (Types.Invalid_local_mutation message) + | Error message -> None, Error (Types.Invalid_local_mutation message) | Ok () -> let id = mutation_id mutation in let fingerprint = fingerprint mutation in (match existing_mutation database id fingerprint with - | Ok (Some existing) -> [], None, Ok (Types.Local_existing existing) - | Error () -> [], None, Error Types.Mutation_identity_conflict + | Ok (Some existing) -> None, Ok (Types.Local_existing existing) + | Error () -> None, Error Types.Mutation_identity_conflict | Ok None -> let context = read_context (snapshot_of_database database) in (match required_preconditions expected mutation with - | Error error -> [], None, Error error + | Error error -> None, Error error | Ok () when not (preconditions_match context expected) -> - [], None, Error Types.Target_precondition_conflict + None, Error Types.Target_precondition_conflict | Ok () -> let intent_time_ms = database.dependencies.epoch_ms () in (match planned_effect database context ~now:intent_time_ms mutation with - | Error error -> [], None, Error error + | Error error -> None, Error error | Ok (effect_footprint, delete_artifacts) -> let candidate_record = local_candidate_record @@ -5676,7 +5676,7 @@ let commit_local database ~expected mutation = mutation in if not (local_candidate_fits database candidate_record) - then [], None, Error Types.Planner_limit_exceeded + then None, Error Types.Planner_limit_exceeded else ( let previous_projection = database.projection in let previous_receipts = database.receipts in @@ -5723,12 +5723,12 @@ let commit_local database ~expected mutation = database.receipts <- previous_receipts; database.outbox <- previous_outbox; database.sync_revision <- previous_sync_revision; - [], None, Error (Types.Local_commit_persistence_failed message) + None, Error (Types.Local_commit_persistence_failed message) | Ok () -> if applied then refresh_queryable_outbox database; database.receipts <- []; if not applied - then [], None, Ok (Types.Local_committed commit) + then None, Ok (Types.Local_committed commit) else ( let change = match summary with @@ -5749,16 +5749,8 @@ let commit_local database ~expected mutation = } | No_logical_change -> assert false in - let callbacks = - database.subscriptions - |> List.filter_map (fun (subscription : subscription) -> - match subscription.lifecycle, subscription.notify with - | Active, Some callback -> Some callback - | Active, None | Unlistened, _ -> None) - in - callbacks, Some change, Ok (Types.Local_committed commit)))))))) + Some change, Ok (Types.Local_committed commit)))))))) in - ignore callbacks; change, outcome) ;; @@ -7644,59 +7636,33 @@ let replan_queued_ordinary database authoritative_database records = Int.compare left.sequence right.sequence) records in - let add_indexed_record ~empty ~append records index select (record : outbox_record) = - ( List.fold_left - (fun index uuid -> - let key = Graph.Uuid.to_string uuid in - let preceding = Uuid_map.find_opt key index |> Option.value ~default:empty in - Uuid_map.add key (append preceding record) index) - index - (select (record.effect_footprint : effect_footprint)) - , record :: records ) - in - let add_record records block_effects page_effects children_effects record = + let add_indexed_record index select (record : outbox_record) = + List.fold_left + (fun index uuid -> + let key = Graph.Uuid.to_string uuid in + let preceding = + Uuid_map.find_opt key index |> Option.value ~default:Rrbvec.empty + in + Uuid_map.add key (Rrbvec.push_back preceding record) index) + index + (select (record.effect_footprint : effect_footprint)) + in + let add_record records block_effects page_effects record = if not (record_is_logically_active record) - then records, block_effects, page_effects, children_effects + then records, block_effects, page_effects else ( - let block_effects, records = - add_indexed_record - ~empty:Rrbvec.empty - ~append:Rrbvec.push_back - records - block_effects - (fun value -> value.block_uuids) - record + let block_effects = + add_indexed_record block_effects (fun value -> value.block_uuids) record in - let page_effects, _ = - add_indexed_record - ~empty:Rrbvec.empty - ~append:Rrbvec.push_back - [] - page_effects - (fun value -> value.page_uuids) - record + let page_effects = + add_indexed_record page_effects (fun value -> value.page_uuids) record in - let children_effects, _ = - add_indexed_record - ~empty:[] - ~append:(fun records record -> records @ [ record ]) - [] - children_effects - (fun value -> - List.filter_map - (function - | Types.Children_interest parent -> Some parent - | Page_tree_interest _ | Journal_index_interest -> None) - value.structure_interests) - record - in - records, block_effects, page_effects, children_effects) + record :: records, block_effects, page_effects) in let rec loop accumulated block_effects page_effects - children_effects unavailable replanned no_change @@ -7706,20 +7672,20 @@ let replan_queued_ordinary database authoritative_database records = | (record : outbox_record) :: rest -> (match record.transport_state, record.mutation with | Types.Queued, (Types.Delete_blocks _ as _mutation) -> - let accumulated, block_effects, page_effects, children_effects = - add_record accumulated block_effects page_effects children_effects record + let accumulated, block_effects, page_effects = + add_record accumulated block_effects page_effects record in loop accumulated block_effects page_effects - children_effects unavailable replanned no_change blocked rest | Types.Queued, mutation -> + (* Replanning only performs point reads; structural roots are not captured. *) let snapshot = { owner = database ; version = @@ -7727,10 +7693,10 @@ let replan_queued_ordinary database authoritative_database records = ; projection_revision = projection_revision database.projection } ; authoritative_database = Some authoritative_database - ; outbox = Some (List.rev accumulated) + ; outbox = None ; block_effects = Some block_effects ; page_effects = Some page_effects - ; children_effects = Some children_effects + ; children_effects = None ; released = false ; active_reads = 0 ; drain_waiters = [] @@ -7742,7 +7708,6 @@ let replan_queued_ordinary database authoritative_database records = accumulated block_effects page_effects - children_effects unavailable replanned (record :: no_change) @@ -7767,7 +7732,6 @@ let replan_queued_ordinary database authoritative_database records = (record :: accumulated) block_effects page_effects - children_effects unavailable replanned no_change @@ -7794,14 +7758,13 @@ let replan_queued_ordinary database authoritative_database records = ; protected_transaction = None } in - let accumulated, block_effects, page_effects, children_effects = - add_record accumulated block_effects page_effects children_effects record + let accumulated, block_effects, page_effects = + add_record accumulated block_effects page_effects record in loop accumulated block_effects page_effects - children_effects unavailable (record :: replanned) no_change @@ -7818,7 +7781,6 @@ let replan_queued_ordinary database authoritative_database records = (record :: accumulated) block_effects page_effects - children_effects unavailable replanned no_change @@ -7828,21 +7790,20 @@ let replan_queued_ordinary database authoritative_database records = | Accepted_pending_authoritative _ | Stale_rejected_pending_authoritative _ ) , _ ) -> - let accumulated, block_effects, page_effects, children_effects = - add_record accumulated block_effects page_effects children_effects record + let accumulated, block_effects, page_effects = + add_record accumulated block_effects page_effects record in loop accumulated block_effects page_effects - children_effects unavailable replanned no_change blocked rest) in - loop [] Uuid_map.empty Uuid_map.empty Uuid_map.empty [] [] [] [] records + loop [] Uuid_map.empty Uuid_map.empty [] [] [] [] records ;; let settle_stale_records ~checkpoint ~origins records = @@ -7979,19 +7940,19 @@ let add_commit_receipts database ~status records receipts = let apply_outbox_transition database preparation ~encrypted = serialize_commit database (fun () -> - let callbacks, change, outcome = + let change, outcome = Eio.Mutex.use_rw ~protect:true database.lock (fun () -> if preparation.transition_owner != database - then [], None, Error Types.Outbox_commit_generation_invalidated + then None, Error Types.Outbox_commit_generation_invalidated else if database.closed - then [], None, Error Types.Outbox_commit_database_closed + then None, Error Types.Outbox_commit_database_closed else if preparation.transition_consumed - then [], None, Error Types.Outbox_preparation_consumed + then None, Error Types.Outbox_preparation_consumed else ( match outbox_submission_batch preparation ~encrypted with - | Error error -> [], None, Error error + | Error error -> None, Error error | Ok _ when database.sync_revision <> preparation.transition_revision -> - [], None, Error Types.Outbox_commit_token_conflict + None, Error Types.Outbox_commit_token_conflict | Ok submission_batch -> preparation.transition_consumed <- true; Option.iter @@ -8000,8 +7961,7 @@ let apply_outbox_transition database preparation ~encrypted = if preparation.plan = Duplicate_plan then ( let revision = projection_revision database.projection in - ( [] - , None + ( None , Ok { Types.generation = database.generation ; before_projection_revision = revision @@ -8280,7 +8240,7 @@ let apply_outbox_transition database preparation ~encrypted = <- stale_earliest_conflict_cursor; record.stale_conflicts <- stale_conflicts) previous_records; - [], None, Error (Types.Outbox_commit_persistence_failed message) + None, Error (Types.Outbox_commit_persistence_failed message) | Ok () -> refresh_queryable_outbox database; database.receipts <- []; @@ -8295,18 +8255,7 @@ let apply_outbox_transition database preparation ~encrypted = ~after_revision:after_projection_revision logical_change_summary in - let callbacks = - match change with - | None -> [] - | Some _ -> - database.subscriptions - |> List.filter_map (fun (subscription : subscription) -> - match subscription.lifecycle, subscription.notify with - | Active, Some callback -> Some callback - | Active, None | Unlistened, _ -> None) - in - ( callbacks - , change + ( change , Ok { Types.generation = database.generation ; before_projection_revision @@ -8318,7 +8267,6 @@ let apply_outbox_transition database preparation ~encrypted = ; submission_batch } )))) in - ignore callbacks; change, outcome) ;; @@ -8744,15 +8692,15 @@ let replay_authoritative database prepared = let commit_authoritative_candidate database prepared = serialize_commit database (fun () -> - let callbacks, change, outcome = + let change, outcome = Eio.Mutex.use_rw ~protect:true database.lock (fun () -> let preparation = prepared.authoritative_preparation in if preparation.authoritative_owner != database - then [], None, Error Types.Authoritative_commit_generation_invalidated + then None, Error Types.Authoritative_commit_generation_invalidated else if database.closed - then [], None, Error Types.Authoritative_commit_database_closed + then None, Error Types.Authoritative_commit_database_closed else if database.sync_revision <> preparation.authoritative_revision - then [], None, Error Types.Authoritative_commit_token_conflict + then None, Error Types.Authoritative_commit_token_conflict else ( match Logseq_db_storage.Storage_session.stage_transact_batch @@ -8761,8 +8709,7 @@ let commit_authoritative_candidate database prepared = prepared.authoritative_transactions with | Error error -> - ( [] - , None + ( None , Error (Types.Authoritative_commit_persistence_failed (storage_session_error error)) ) @@ -9196,8 +9143,7 @@ let commit_authoritative_candidate database prepared = (List.map Persistence_receipt_v1.encode_terminal_batch terminal_batches) with | Error error -> - ( [] - , None + ( None , Error (Types.Authoritative_commit_persistence_failed (storage_session_error error)) ) @@ -9210,7 +9156,7 @@ let commit_authoritative_candidate database prepared = (match replay_authoritative database prepared with | Error message -> database.closed <- true; - [], None, Error (Types.Authoritative_commit_fatal_state message) + None, Error (Types.Authoritative_commit_fatal_state message) | Ok () -> database.outbox <- candidate_outbox; database.receipts <- []; @@ -9271,18 +9217,7 @@ let commit_authoritative_candidate database prepared = ~after_revision logical_change_summary in - let callbacks = - match change with - | None -> [] - | Some _ -> - database.subscriptions - |> List.filter_map (fun (subscription : subscription) -> - match subscription.lifecycle, subscription.notify with - | Active, Some callback -> Some callback - | Active, None | Unlistened, _ -> None) - in - ( callbacks - , change + ( change , Ok { Types.generation = database.generation ; before_projection_revision = before_revision @@ -9305,7 +9240,6 @@ let commit_authoritative_candidate database prepared = ; logical_change_summary } ))))) in - ignore callbacks; change, outcome) ;;