feat(big_repo): clean rewrite - #33
Conversation
📝 WalkthroughWalkthroughThis change replaces the legacy BigRepo runtime with runtime2, adds SQLite-backed persistence, generalizes transport handlers, records Keyhive content frontiers, updates sync and notification behavior, and adds broad integration coverage for access, convergence, encryption, restart, topology, and error scenarios. ChangesRuntime2 migration and repository persistence
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
🟠 Major comments (39)
src/big_repo/handler.rs-216-224 (1)
216-224: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPropagate ephemeral handler failures instead of reporting success.
Both implementations log the error and return
Ok(()), silently dropping failed protocol messages. Add an ephemeral variant toBigRepoComposedHandlerErrorand return it fromhandle.As per coding guidelines,
**/*.{rs,kt,tsx?,jsx?}must not swallow errors with catch-and-log unless explicitly requested.Also applies to: 275-283
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/handler.rs` around lines 216 - 224, Update the BigRepoWireMessage::Ephemeral handling in handle to propagate failures instead of logging them and returning Ok(()). Add an ephemeral-handler variant to BigRepoComposedHandlerError, map the Handler::<Sendable, C>::handle error into that variant, and return it from both affected implementations while preserving successful handling.Source: Coding guidelines
src/big_repo/runtime.rs-1338-1342 (1)
1338-1342: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winTrack keyhive completions by
request_id.
finish_keyhive_syncignores the completion’srequest_id, so a stale or duplicate event for the peer can clear the active round and unblock waiters early. Store the expectedrequest_idwith the active sync and drop mismatches.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime.rs` around lines 1338 - 1342, Update the RuntimeEvt::KeyhiveSyncDone handling and finish_keyhive_sync flow to retain the active sync’s expected request_id, compare each completion event against it, and ignore stale or duplicate mismatches without clearing state or waking waiters; only matching completions should finish the sync.src/big_repo/changes.rs-1284-1288 (1)
1284-1288: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winMatch the full affected sequence range.
Insertshifts every listener at or after the insertion point, andDeleteSeqremoves a span of elements; comparing only the start index misses listeners on affected later positions.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/changes.rs` around lines 1284 - 1288, Update the sequence-index matching branch for PutSeq, Insert, and DeleteSeq to evaluate the complete affected range rather than only comparing listener_idx with the operation’s start index. Treat Insert as affecting listeners at or after index, and DeleteSeq as affecting the span removed by the action; preserve the existing PutSeq behavior for its affected position.src/big_repo/test2/harness/topo.rs-227-231 (1)
227-231: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDo not conceal teardown invariant failures.
Both paths silently leave teardown incomplete, allowing leaked workers or endpoints to surface later as unrelated test flakes.
src/big_repo/test2/harness/topo.rs#L227-L231: propagate orexpectboth stop results.src/big_repo/test2/harness/topo.rs#L275-L279: treat dropping outside the expected Tokio runtime as an invariant failure instead of leaking nodes.As per coding guidelines, “Don't ignore channel send errors in Rust,” and unexpected invariant failures should be unwrapped.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/harness/topo.rs` around lines 227 - 231, Update shutdown in src/big_repo/test2/harness/topo.rs:227-231 to propagate or explicitly expect both repo_stop.stop() and big_sync_stop.stop() results instead of ignoring errors. At src/big_repo/test2/harness/topo.rs:275-279, make dropping outside the expected Tokio runtime fail explicitly by unwrapping or expecting the invariant result; do not allow either teardown failure to be silently ignored.Source: Coding guidelines
src/big_repo/test2/harness/topo.rs-304-318 (1)
304-318: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftInstall the shutdown guard before subsequent fallible setup.
Each builder creates nodes, connections, or workers before ownership reaches
ShutdownGuard. An error during later setup drops bareNodevalues without running their asynchronous shutdown, contradicting the setup-failure guarantee.
src/big_repo/test2/harness/topo.rs#L304-L318: guard the left node before booting the right node.src/big_repo/test2/harness/topo.rs#L321-L351: guard each persistent node before connecting or syncing.src/big_repo/test2/harness/topo.rs#L507-L532: build the relay topology inside an incrementally populated guard.src/big_repo/test2/harness/topo.rs#L536-L558: build the line topology inside an incrementally populated guard.src/big_repo/test2/harness/topo.rs#L563-L586: build the star topology inside an incrementally populated guard.src/big_repo/test2/harness/topo.rs#L591-L619: build the triangle topology inside an incrementally populated guard.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/harness/topo.rs` around lines 304 - 318, Install ShutdownGuard incrementally before any subsequent fallible setup so already-booted nodes receive asynchronous shutdown on errors. In boot_disconnected, guard the left node before booting the right; in the persistent-node builder, guard each node before connecting or syncing; and in the relay, line, star, and triangle topology builders, populate the guard as each node is created before continuing setup. Apply these changes at src/big_repo/test2/harness/topo.rs ranges 304-318, 321-351, 507-532, 536-558, 563-586, and 591-619, respectively.src/big_repo/test2/harness/fixtures.rs-151-160 (1)
151-160: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not mask the synchronous sync barrier with quiescence.
The post-sync
wait_for_quiescenceallows delayed materialization to complete before the assertion, so this helper cannot detect the premature-return bug described by its contract. Checkget_docimmediately, or rename the helper and test the barrier separately.Proposed fix
conn.sync_doc_with_peer(doc_id, Some(std::time::Duration::from_secs(10))) .await?; -repo.wait_for_quiescence(Some(std::time::Duration::from_secs(10))) - .await?; match repo.get_doc(&doc_id).await? {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/harness/fixtures.rs` around lines 151 - 160, Remove the post-sync repo.wait_for_quiescence call from sync_doc_expect_ready so get_doc is checked immediately after sync_doc_with_peer returns. Preserve the existing 10-second timeout on the synchronous sync call and the helper’s current get_doc result handling.src/big_repo/test2/harness/topo.rs-80-100 (1)
80-100: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReopen the SQLite store for persistent restarts.
boot_with_configalways createsSqliteBigRepoStorewithSqlCtx::memory(), andrestartjust reuses the sameArc<SqliteBigRepoStore>. That meansboot_persistentnever exercises a disk-backed reopen, so restart coverage can pass without loading state from SQLite.src/big_repo/test2/harness/topo.rs:80-100,159-165🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/harness/topo.rs` around lines 80 - 100, The persistent test setup around SqliteBigRepoStore::new and restart must use a file-backed SQLite context instead of SqlCtx::memory(), and restart must reopen a new store from that same database path rather than reuse the existing Arc. Preserve the existing initialization and boot flow while ensuring boot_persistent validates state reload across a genuine SQLite close-and-reopen.src/big_repo/test2/revocation.rs-76-110 (1)
76-110: 🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy liftVerify forward secrecy with post-revocation content.
Checking
agent_access_on(...).is_none()only validates metadata. Write new owner content after revocation, attempt synchronization, and assert the retained reader handle cannot observe it; otherwise continued key delivery would pass this test.As per coding guidelines, shallow tests should be proactively strengthened.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/revocation.rs` around lines 76 - 110, Strengthen the revocation test around the existing agent_access_on assertion by having the owner write distinct document content after revocation, attempt the normal synchronization flow, and then read through reader_doc. Assert that the retained reader handle cannot observe the post-revocation content, while preserving the existing assertion that previously held plaintext remains readable.Source: Coding guidelines
src/big_repo/test2/notifications.rs-104-112 (1)
104-112: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winWait for runtime quiescence before asserting notification absence.
These immediate
try_recv()checks only prove that nothing is queued yet. A broken filter can deliver asynchronously after the assertion and still pass the test.Add a deterministic barrier at each negative assertion
+ pair.left().repo.wait_for_quiescence(None).await?; assert_no_notification(&mut rx);As per coding guidelines, tests should be proactively strengthened when their assertions are too shallow.
Also applies to: 166-174, 270-278
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/notifications.rs` around lines 104 - 112, Strengthen the negative notification checks by waiting for runtime quiescence before asserting no notification is received. Update assert_no_notification and each corresponding call site, including the cases around the doc B write and the additionally referenced assertions, so asynchronous deliveries are fully settled before the absence check while preserving the existing filtered-document test behavior.Source: Coding guidelines
src/big_repo/test2/cgka.rs-306-327 (1)
306-327: 🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy liftAssert newly emitted CGKA semantics instead of aggregate counts.
These checks permit false positives:
- Multipath dedup passes when the second path emits one duplicate operation.
- Group checkpoint passes even if member addition emits nothing.
- Structural Add versus checkpoint Update never examines operation variants.
Capture the baseline hashes, inspect only newly introduced operations, and assert their payload kinds, principals, and frontiers.
As per coding guidelines, “See a useless or too shallow test? Flag it for removal.”
Also applies to: 454-474, 525-575
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/cgka.rs` around lines 306 - 327, Replace the aggregate CGKA operation-count assertions in the multipath, group-checkpoint, and structural Add/Update tests with semantic inspection of newly emitted operations. Capture baseline operation hashes before each action, isolate only new operations afterward, and assert their payload variants, principals, and frontiers; ensure the checkpoint test verifies an expected member-add operation rather than merely observing a count change. Remove any test that cannot make these meaningful semantic assertions.Source: Coding guidelines
src/utils_rs/lib.rs-845-856 (1)
845-856: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winTreat cancelled
JoinErrors as expected cleanup instop()
abort_all()leaves tasks in the set, sojoin_next().awaitwill surfaceis_cancelled()errors for ordinary shutdown.out?turns the documentedabort()→stop()flow into a failure path; skip cancelled joins and only propagate unexpected errors.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/utils_rs/lib.rs` around lines 845 - 856, Update stop() to ignore JoinError values where is_cancelled() is true while draining tasks aborted by abort(). Continue joining all remaining tasks and propagate only unexpected join failures, preserving the existing abort() → stop() cleanup flow.src/big_repo/test2/topologies.rs-125-143 (1)
125-143: 🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy liftDo not use public Read access in Relay-only topology tests.
Public Read also authorizes the intermediate relay, invalidating the assertion that it remains
Access::Relay.
src/big_repo/test2/topologies.rs#L125-L143: propagate Bob’s agent identity back to Alice and grant Bob directly while retaining Relay-only access for R.src/big_repo/test2/topologies.rs#L252-L268: propagate Carol’s identity back through Bob and grant Carol directly rather than granting public Read.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/topologies.rs` around lines 125 - 143, The Relay-only topology tests must avoid public Read grants, which also authorize the relay. At src/big_repo/test2/topologies.rs lines 125-143, propagate Bob’s agent identity back to Alice and grant Bob direct Read access while preserving the relay’s Access::Relay capability; at lines 252-268, propagate Carol’s identity through Bob and grant Carol directly instead of using public Read.src/big_repo/test2/restart.rs-576-597 (1)
576-597: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDrop pre-restart handles and reacquire them from the restarted left repository.
Both tests retain
owner_doc, which references the old runtime and holds its lease, then use that stale handle in post-restart invariants.
src/big_repo/test2/restart.rs#L576-L597: dropowner_docbeforerestart_left, then obtain a fresh left-side handle after reconnect.src/big_repo/test2/restart.rs#L643-L668: drop the writer handle after the awaited local write, restart, and verify persistence using a newly acquired owner handle.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/restart.rs` around lines 576 - 597, In src/big_repo/test2/restart.rs:576-597, drop owner_doc before restart_left, then reacquire a fresh left-side owner handle after reconnect and use it for the post-restart heads::tier0_invariants call; update cleanup to drop the replacement handle. In src/big_repo/test2/restart.rs:643-668, drop the writer handle after the awaited local write, restart, and acquire a new owner handle before verifying persistence. Preserve the existing restart and synchronization flow.src/big_repo/test2/capability.rs-475-498 (1)
475-498: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winMake the authorization tests fail only for authorization violations.
These cases can currently pass without exercising the intended permission boundary.
src/big_repo/test2/capability.rs#L475-L498: invoke the Edit grant from the actual Read-only node, not an unrelated agent with no document; assert an authorization-specific error and inspect that actor’s state.src/big_repo/test2/capability.rs#L573-L593: materialize the document for ReaderA before revocation, assert an authorization-specific error, and compare ReaderA’s local state rather than the unsynchronized Owner state.src/big_repo/test2/capability.rs#L1692-L1737: useexpect_errfor both attempts and requireobs_access.is_none(); the current condition explicitly permits forbiddenSome(Access::Edit).As per coding guidelines, proactively improve shallow tests and let unexpected Rust errors fail.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/capability.rs` around lines 475 - 498, Strengthen the authorization tests in src/big_repo/test2/capability.rs:475-498, 573-593, and 1692-1737. In the first site, invoke the Edit grant through the actual read-only node, require an authorization-specific error, and compare that actor’s document state; in the second, materialize the document for ReaderA before revocation, require an authorization-specific error, and compare ReaderA’s local state; in the third, use expect_err for both attempts and require obs_access.is_none(), rejecting forbidden Some(Access::Edit) results. Let unexpected Rust errors propagate normally.Source: Coding guidelines
src/big_repo/test2/convergence.rs-53-57 (1)
53-57: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winBound every quiescence barrier with a timeout.
A runtime regression can otherwise hang the test process indefinitely.
src/big_repo/test2/convergence.rs#L53-L57: replaceNonewithSome(Duration::from_secs(10)).src/big_repo/test2/convergence.rs#L253-L257: replaceNonewith the existing ten-second timeout convention.src/big_repo/test2/convergence.rs#L308-L316: bound both repository waits.src/big_repo/test2/edge.rs#L175-L180: bound the authorized-reader readiness wait.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/convergence.rs` around lines 53 - 57, Bound every quiescence/readiness wait with the existing ten-second timeout convention: update src/big_repo/test2/convergence.rs lines 53-57 and 253-257 to pass Some(Duration::from_secs(10)) instead of None, update both repository waits in lines 308-316, and apply the same timeout to the authorized-reader readiness wait in src/big_repo/test2/edge.rs lines 175-180.src/big_repo/test2/encryption.rs-398-442 (1)
398-442: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winEnsure the test actually identifies a post-revocation blob.
post_revoke_undecryptable_by_revokedstarts astrue, so the test passes if no decrypted owner blob containspostrevoke. Track and assert that the target blob was found.Proposed fix
let mut pre_revoke_decryptable = false; + let mut found_post_revoke_blob = false; let mut post_revoke_undecryptable_by_revoked = true; @@ { + found_post_revoke_blob = true; if revoked_result.is_ok() { post_revoke_undecryptable_by_revoked = false; } @@ assert!( + found_post_revoke_blob, + "must identify the post-revoke blob before testing forward secrecy" + ); + assert!( post_revoke_undecryptable_by_revoked,🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/encryption.rs` around lines 398 - 442, Update the post-revocation verification in the blob loop to track whether an owner-decryptable blob containing “postrevoke” was found, separately from whether the revoked peer could decrypt it. Assert that the target post-revocation blob was found, while preserving the existing assertion that the revoked peer cannot decrypt it.Source: Coding guidelines
src/big_repo/test2/capability.rs-1881-1922 (1)
1881-1922: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSync Charlie’s delegated grants back to Owner before asserting Owner’s access view.
Only Alice↔Charlie and Bob↔Charlie are synchronized. Line 1912 therefore checks Owner before any path returns Charlie’s merged delegation state.
Proposed fix
bob_charlie.sync_keyhive_with_peer(None).await?; charlie_bob.sync_keyhive_with_peer(None).await?; + + owner_charlie.sync_keyhive_with_peer(None).await?; + charlie_owner.sync_keyhive_with_peer(None).await?; // Charlie's own keyhive must show Read access after receiving both paths.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/capability.rs` around lines 1881 - 1922, Synchronize Charlie’s merged delegation state back to the owner before evaluating owner_sees: after the Alice↔Charlie and Bob↔Charlie exchanges, use the existing peer sync flow to sync Charlie with the owner in both directions, then keep the owner access assertion unchanged.src/big_repo/runtime2/hub.rs-305-318 (1)
305-318: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftTrack spawned finite work before resolving quiescence.
CreateDoccan still be running when a followingWaitForQuiescenceobserves no workers or keyhive rounds and resolves immediately. Add in-flight child-work accounting to the probe condition, incremented before spawn and released by completion events.Also applies to: 340-354, 532-536
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/hub.rs` around lines 305 - 318, Update try_resolve_quiescence and the CreateDoc spawn/completion flow to track finite child work in the quiescence probe: increment the in-flight count before spawning CreateDoc work, and decrement it when the corresponding completion event is handled. Include this count in the probe’s early-return condition so WaitForQuiescence cannot resolve while spawned child work remains active.src/big_repo/runtime2/hub.rs-1315-1341 (1)
1315-1341: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDo not downgrade actor lifecycle failures to warnings.
These branches silently drop materialization retries, internal keyhive syncs, or sync cancellations. Treat worker lookup/send and task-spawn failures as invariant violations using
expect,unwrap, or propagated errors.As per coding guidelines, Rust errors must not be swallowed with catch-and-log handling.
Also applies to: 1384-1394
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/hub.rs` around lines 1315 - 1341, Promote actor lifecycle failures to invariant violations in reattempt_pending_materialization, schedule_internal_keyhive_sync, and the related sync-cancellation logic around the referenced code. Replace worker lookup and task-spawn catch-and-log branches with expect/unwrap or propagated errors, ensuring materialization retries, internal syncs, and cancellations are never silently dropped; retain only intentional state handling such as dirty-sync tracking.Source: Coding guidelines
src/big_repo/runtime2/hub.rs-207-212 (1)
207-212: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not report storage failures as “not resident”.
unwrap_or(false)makes SQLite/IO failure indistinguishable from absent content. Preserve the error or panic according to the runtime invariant policy.Proposed fix
let stored = runtime_io .contains_sedimentree(sed_id) .await - .unwrap_or(false); -let _ = resp.send(stored); + .expect("failed checking sedimentree residency"); +resp.send(stored) + .expect("sedimentree residency receiver must remain open");As per coding guidelines, unexpected errors should crash rather than be concealed by fallback values.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/hub.rs` around lines 207 - 212, Update the async flow around contains_sedimentree in F::from_future so storage errors are not converted to false via unwrap_or(false). Propagate the error or trigger the runtime’s established failure behavior, while retaining the existing response send for successful presence checks.Source: Coding guidelines
src/big_repo/runtime2/hub.rs-805-827 (1)
805-827: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winClose the accepted connection if its watcher cannot be spawned.
Unlike the outbound path, this error branch leaves the authenticated inbound transport open and untracked.
Proposed fix
if let Err(error) = watcher { + connect.close(handshake_peer).await?; let _ = resp.send(Err(error)); return Ok(()); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/hub.rs` around lines 805 - 827, Update the watcher-spawn error branch around child_tasks.spawn and the accepted inbound transport so that a failed watcher creation closes the authenticated connection before returning the error response. Preserve the existing resp.send(Err(error)) and return flow, ensuring the transport is not left open or untracked.src/big_repo/runtime2/hub.rs-851-862 (1)
851-862: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPropagate fire-and-forget connection close failures.
When
respisNone, teardown errors are discarded after the hub has already removed the peer, potentially leaving an untracked live connection.Proposed fix
let result = connect.close(peer_id).await; -if let Some(resp) = resp { - let _ = resp.send(result); +match resp { + Some(resp) => { + resp.send(result) + .expect("connection close receiver must remain open"); + Ok(()) + } + None => result, } -Ok(())As per coding guidelines, unexpected task errors must not be swallowed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/hub.rs` around lines 851 - 862, Update close_connection_async so the result of connect.close(peer_id).await is propagated when resp is None instead of always returning Ok(()). Preserve sending the result through resp when present, while returning the close error from the F::Future path for fire-and-forget calls.Source: Coding guidelines
src/big_repo/runtime2/doc_worker.rs-660-685 (1)
660-685: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not suppress patches when heads are unchanged.
The early return occurs before the patch loop, contradicting the stated tombstone-compaction behavior. Skip only head notifications; still emit
notify_doc_changedfor every patch.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/doc_worker.rs` around lines 660 - 685, The early return in notify_heads_changed incorrectly suppresses patch notifications when heads are unchanged. Restrict the last_notified_heads check to notify_doc_heads_changed, while always executing the existing patches loop so notify_doc_changed is emitted for every patch.src/big_repo/runtime2/doc_worker.rs-1670-1688 (1)
1670-1688: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDo not clear peer-wide completion state when cancelling one waiter.
A late waiter timing out removes
sync_completedandsync_materializedfor the active round. If one completion half already arrived, existing waiters can then remain unresolved forever. Clear these flags only when cancelling the active round itself.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/doc_worker.rs` around lines 1670 - 1688, The waiter-specific cancellation branch in the pending_sync_jobs cleanup must not remove peer-wide sync_completed or sync_materialized state. Move those removals so they occur only when waiter_id is None, while preserving active_doc_syncs and pending waiter cleanup for individual waiter cancellation.src/big_repo/lib.rs-480-483 (1)
480-483: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftPreserve the connection-finish callback contract.
Both public APIs retain
end_signal_txbut discard it, so callers waiting forConnFinishSignalare never notified when the runtime2 connection ends. Wire the sender into the connection end future or remove it through an explicit breaking API change.Also applies to: 501-502
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/lib.rs` around lines 480 - 483, Update the connection setup around runtime.open_connection in both affected public APIs to use end_signal_tx for the connection-finish notification instead of discarding it. Wire the sender into the connection end future so callers receive ConnFinishSignal when the connection closes; do not remove the parameter or make an unrelated API change.src/big_repo/runtime2/io.rs-203-209 (1)
203-209: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftPreserve application keys across a commit batch.
The native implementation passes a fresh empty batch-key map and discards each returned application key. A child commit encrypted in the same delta therefore cannot include its newly encrypted parent key in the causal envelope. Change this seam to encrypt the batch or return/thread the application key.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/io.rs` around lines 203 - 209, Update the encrypt_loose_commit seam and its native implementation to preserve and thread the batch application-key map across all commits in a delta, rather than creating an empty map per commit and discarding returned keys. Ensure each child commit can include the newly encrypted parent key in its causal envelope, while retaining the existing EncryptedLooseCommit result behavior.src/big_repo/runtime2/doc_worker.rs-575-589 (1)
575-589: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winEmit every generated CGKA operation.
All operations are persisted, but only
cgka_ops.into_iter().next()reaches the hub. Later epoch updates are silently omitted from runtime event processing.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/doc_worker.rs` around lines 575 - 589, Update the CGKA event emission in the operation-persistence flow to send one Runtime2Evt::CgkaOp event for every entry in cgka_ops, not only the first operation. Preserve persistence of all operations and propagate channel-send errors for each emitted event.src/big_repo/runtime2/doc_worker.rs-272-279 (1)
272-279: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftPersist CGKA metadata before making ciphertext durable.
Both paths store encrypted content before persisting the generated CGKA operations. If CGKA persistence fails, durable ciphertext can reference an epoch that cannot be reconstructed after restart. Persist the operations first, idempotently, or transactionally couple both writes.
Also applies to: 561-579
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/doc_worker.rs` around lines 272 - 279, Update both persistence paths around store_initial_sedimentree and persist_cgka_update_op so all generated CGKA operations are persisted before encrypted content becomes durable. Preserve idempotency or use transactional coupling between the writes, and apply the same ordering to the additional path around the referenced CGKA handling.src/big_repo/sqlite_big_repo_store.rs-1369-1405 (1)
1369-1405: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winUpsert the added member instead of deleting every member.
add_doc_memberclears the entire document ACL in SQL before inserting one principal. The in-memory cache retains the others, producing immediate SQL/cache divergence and losing them after restart.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/sqlite_big_repo_store.rs` around lines 1369 - 1405, Update add_doc_member to upsert only the specified member’s ACL row, matching on scope_id, obj_id, and principal_id; remove the statement that deletes all rows for the document. Preserve existing members in SQL and keep the cache update for the added member.src/big_repo/runtime2/doc_worker.rs-256-258 (1)
256-258: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winHandle request-response channel closure explicitly.
These oneshot sends are not broadcasts; discarding their errors hides lifecycle violations. Unwrap/propagate them, or explicitly model caller cancellation where it is expected.
As per coding guidelines, “Don't ignore channel send errors in Rust.”
Also applies to: 320-321, 385-386, 592-593, 871-876
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/doc_worker.rs` around lines 256 - 258, Update the response sends in the document worker, including the branches around the occupied-state check and the listed additional locations, to handle oneshot channel closure explicitly instead of discarding send errors with `let _ =`. Propagate or unwrap each send failure according to the surrounding function’s error-handling conventions, preserving the existing response and return behavior.Source: Coding guidelines
src/big_repo/sqlite_big_repo_store.rs-1067-1109 (1)
1067-1109: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAdvance
big_sync_parts.latest_cursoron removal.The member and bucket cursors advance, but the part summary remains stale. Peers using
PartSummary.latest_cursorcan conclude that no removal occurred.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/sqlite_big_repo_store.rs` around lines 1067 - 1109, The removal transaction must also advance the containing part’s latest cursor. In the flow after obtaining cursor in the member-removal logic, update big_sync_parts.latest_cursor for self.scope_id and part_id using the same cursor value, while preserving the existing member, bucket, object, and commit behavior.src/big_repo/sqlite_big_repo_store.rs-1250-1266 (1)
1250-1266: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReturn the last delivered cursor, not the first omitted cursor.
Subsequent queries use
latest_cursor > cursor. Settingnext_cursorto the first omitted event causes that event to be skipped on the next page.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/sqlite_big_repo_store.rs` around lines 1250 - 1266, Update the pagination logic around events sorting and next_cursor so PartPage.next_cursor contains the cursor of the last event delivered, not the first omitted event. Preserve the existing limit-based truncation and leave next_cursor unset when no additional events exist; use the delivered page’s final event after applying the limit.src/big_repo/sqlite_big_repo_store.rs-1332-1366 (1)
1332-1366: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winDo not replace complete ACLs with the boot-time partial seed.
BigRepo::boot_innercalls this with only the local principal, but this method deletes every persisted principal and replaces the cache. Restarting therefore removes all other document members. Pass the complete ACL or provide a merge/upsert operation for boot seeding.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/sqlite_big_repo_store.rs` around lines 1332 - 1366, Update the boot seeding flow around BigRepo::boot_inner and set_doc_members so boot-time initialization cannot overwrite persisted document members with only the local principal. Pass the complete ACL to set_doc_members, or add and use a merge/upsert operation that preserves existing principals in both big_sync_syncable and doc_members_cache while applying the local seed.src/big_repo/runtime2/native.rs-1396-1435 (1)
1396-1435: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPropagate unexpected background-task failures.
The listener ignores its inner result, the manager converts errors with
.ok(), and maintenance failures are caught and logged indefinitely. These tasks have no failure-reporting channel, so unexpected failures should terminate throughunwrap().As per coding guidelines, unexpected Tokio task errors should panic and errors must not be swallowed with catch-and-log.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/native.rs` around lines 1396 - 1435, Update the background tasks spawned around the listener, manager, and Keyhive maintenance loop to propagate unexpected failures by unwrapping task and operation results instead of ignoring them, converting them with .ok(), or catching them with map_err logging. Preserve the existing maintenance timing and retry loop, but ensure listener.await, manager.await, refresh_cache(), and compact() failures terminate through unwrap().Source: Coding guidelines
src/big_repo/sqlite_big_repo_store.rs-1272-1316 (1)
1272-1316: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftClose the replay-to-subscription race.
The subscriber is registered only after querying and sending the replay. Any event committed between the replay query and bus registration is permanently missed. Register a buffered subscription before taking the replay snapshot, then deduplicate by cursor.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/sqlite_big_repo_store.rs` around lines 1272 - 1316, The subscribe flow in subscribe currently registers with the bus only after replay delivery, allowing committed events to be missed. Create and register a buffered subscription before taking the replay snapshot, then replay events while deduplicating against each requested part’s cursor so events already included in the snapshot are not delivered twice.src/big_repo/sqlite_big_repo_store.rs-1817-1847 (1)
1817-1847: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftUpdate the big-sync payload when sedimentree data is deleted.
All save paths recompute
payload_json, but deletion paths leave the old frontier advertised. Deleting a head or the full sedimentree can therefore leave peers synchronizing against nonexistent content. Recompute or remove the payload in the same transaction and publish the resulting event.Also applies to: 1967-1990, 2067-2089
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/sqlite_big_repo_store.rs` around lines 1817 - 1847, Update delete_sedimentree_id and the corresponding deletion paths around the referenced symbols to recompute or remove the big-sync payload within the same database transaction after deleting sedimentree data, then publish the resulting payload event. Ensure deleting either a head or an entire sedimentree cannot leave the previous frontier advertised, while preserving the existing deletion behavior.src/big_repo/lib.rs-374-383 (1)
374-383: 🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy liftSerialize access changes with document commit encryption.
These methods snapshot the sedimentree frontier and mutate Keyhive in separate asynchronous steps while document writes can continue. The supplied
after_contentcan become stale; for revocation, an old-epoch commit can race past the revocation frontier. Route access changes through the document actor or hold a write barrier across the snapshot and Keyhive operation.Also applies to: 415-426, 450-460
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/lib.rs` around lines 374 - 383, Serialize the access-change flow around the sedimentree snapshot and Keyhive mutation: update the method containing the after_content construction and the related paths at the noted locations to run through the document actor or hold the document write barrier across both doc_head_state and the Keyhive add/remove operation. Ensure document commits cannot occur between frontier capture and completion of the access change, preserving a current revocation frontier for encryption.src/big_repo/runtime2/native.rs-277-300 (1)
277-300: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winTreat a missing Keyhive document as pending materialization.
DocIodefines unavailable keys asNone, butget_kh_docturns a not-yet-synced document into an error. Sedimentree content arriving before Keyhive state therefore fails lookup/materialization instead of enteringPendingMaterialization.Also applies to: 596-610, 663-680
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/native.rs` around lines 277 - 300, Update get_kh_doc and its callers to represent a locally missing Keyhive document as unavailable materialization state rather than an error: return the None-compatible result expected by DocIo, and propagate that through the lookup/materialization paths around the referenced ranges so sedimentree content enters PendingMaterialization. Preserve errors for actual failures and retain normal document handling when the document is present.src/big_repo/runtime2/native.rs-238-252 (1)
238-252: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftImplement PCS-update ciphertext lookup
src/big_repo/runtime2/native.rs:238-252is used bytry_causal_decrypt, so returningVec::new()here causes causal walks to miss ciphertexts that are only reachable through their CGKA update. Mirror the reversepcs_updateslookup fromsrc/big_repo/runtime.rs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/native.rs` around lines 238 - 252, Implement get_ciphertext_by_pcs_update by mirroring the pcs_updates lookup in runtime.rs, returning the encrypted content associated with the supplied CGKA update digest instead of an empty vector. Preserve the existing FutureForm and error-result signature so try_causal_decrypt can retrieve ciphertexts during causal walks.
🟡 Minor comments (4)
src/big_repo/test2/harness/fixtures.rs-138-142 (1)
138-142: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winReport the document identifier in access failures.
The current message renders
"Alice has no access on Alice", obscuring which document failed.Proposed fix
Err(crate::ferr!( - "{} has no access on {} after grant + keyhive sync", + "{} has no access on {doc_id} after grant + keyhive sync", log_nickname::nickname(&peer), - log_nickname::nickname(&peer), // doc_id has no nickname; peer stands in ))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/harness/fixtures.rs` around lines 138 - 142, The access-failure error in the relevant test harness assertion currently uses the peer nickname for both placeholders. Replace the second argument with the document identifier variable so the message identifies the failing document while retaining the peer nickname as the actor.src/big_repo/test2/harness/topo.rs-62-63 (1)
62-63: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winSwitch this handoff to
notify_one().accepted_connection()can slip between the slot check andnotified().await, andnotify_waiters()doesn’t store a permit, so the accepted connection can sit in the slot until the timeout. This is a one-consumer path, sonotify_one()(or anmpscchannel) fits better.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/harness/topo.rs` around lines 62 - 63, Change the notification in the connection handoff after updating self.accepted from notify_waiters() to notify_one(), preserving the existing single-consumer accepted_connection() flow.src/big_repo/test2/notifications.rs-637-670 (1)
637-670: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winFail when the listener channel disconnects.
_regis still held here, soDisconnectedshould not happen; breaking early can let the test pass after only partial delivery.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/notifications.rs` around lines 637 - 670, Update the rx.try_recv loop to fail the test when it returns TryRecvError::Disconnected instead of breaking. Keep TryRecvError::Empty as the normal termination condition, ensuring the test cannot pass after partial notification delivery while _reg remains held.src/big_repo/test2/edge.rs-197-215 (1)
197-215: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winRequire
NotFoundinstead of accepting a successful unauthorized sync.The stated contract is
SyncDocError::NotFound, but this branch acceptsOk(()). Enforce the error and retain the no-materialization assertion.Proposed fix
- let intruder_sync = intruder_owner_conn + let intruder_error = intruder_owner_conn .sync_doc_with_peer(doc_id, Some(Duration::from_secs(10))) - .await; - match intruder_sync { - Ok(()) => { - let lookup = guard.node(2).repo.get_doc(&doc_id).await?; - assert!( - !matches!(lookup, crate::DocLookup::Ready(_)), - "unauthorized peer must not materialise plaintext" - ); - } - Err(err) => { - assert!( - matches!(err, SyncDocError::NotFound), - "unauthorized doc sync should fail with NotFound, got {err:?}" - ); - } - } + .await + .expect_err("unauthorized doc sync must fail"); + assert!(matches!(intruder_error, SyncDocError::NotFound)); + + let lookup = guard.node(2).repo.get_doc(&doc_id).await?; + assert!(!matches!(lookup, crate::DocLookup::Ready(_)));🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/edge.rs` around lines 197 - 215, Update the unauthorized sync assertion around intruder_owner_conn.sync_doc_with_peer so it must return SyncDocError::NotFound rather than accepting Ok(()). Preserve the existing assertion that the unauthorized document does not materialise plaintext, applying it after the expected error if the test contract requires retaining that check.
🧹 Nitpick comments (1)
src/daybook_ffi/listener_bridge.rs (1)
1-2: 🚀 Performance & Scalability | 🔵 TrivialTrack the thread-per-listener scalability work.
Would you like me to open an issue for replacing this with a shared executor or dispatcher?
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/daybook_ffi/listener_bridge.rs` around lines 1 - 2, Track the thread-per-listener scalability concern by creating a follow-up issue for the listener implementation in listener_bridge.rs, covering replacement of per-listener threads with a shared executor or dispatcher; do not change the current implementation in this fix.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: eda7bd6d-c50b-40c0-9051-509f698fa368
📒 Files selected for processing (48)
src/big_repo/Cargo.tomlsrc/big_repo/backend.rssrc/big_repo/changes.rssrc/big_repo/ephemeral.rssrc/big_repo/handler.rssrc/big_repo/keyhive.rssrc/big_repo/keyhive_conn.rssrc/big_repo/lib.rssrc/big_repo/runtime.rssrc/big_repo/runtime2/doc_worker.rssrc/big_repo/runtime2/handle.rssrc/big_repo/runtime2/hub.rssrc/big_repo/runtime2/io.rssrc/big_repo/runtime2/lease.rssrc/big_repo/runtime2/messages.rssrc/big_repo/runtime2/mod.rssrc/big_repo/runtime2/native.rssrc/big_repo/runtime2/tasks.rssrc/big_repo/runtime2/test_support.rssrc/big_repo/sqlite_big_repo_store.rssrc/big_repo/test.rssrc/big_repo/test2/access_matrix.rssrc/big_repo/test2/capability.rssrc/big_repo/test2/cgka.rssrc/big_repo/test2/convergence.rssrc/big_repo/test2/edge.rssrc/big_repo/test2/encryption.rssrc/big_repo/test2/harness/dump.rssrc/big_repo/test2/harness/fixtures.rssrc/big_repo/test2/harness/heads.rssrc/big_repo/test2/harness/keyhive.rssrc/big_repo/test2/harness/log_nickname.rssrc/big_repo/test2/harness/mod.rssrc/big_repo/test2/harness/topo.rssrc/big_repo/test2/ladder.rssrc/big_repo/test2/mod.rssrc/big_repo/test2/notifications.rssrc/big_repo/test2/restart.rssrc/big_repo/test2/revocation.rssrc/big_repo/test2/topologies.rssrc/big_sync/lib.rssrc/big_sync/part_store.rssrc/big_sync_core/ids.rssrc/daybook_core/repo.rssrc/daybook_core/sync.rssrc/daybook_core/sync/bootstrap.rssrc/daybook_ffi/listener_bridge.rssrc/utils_rs/lib.rs
💤 Files with no reviewable changes (1)
- src/big_sync_core/ids.rs
| pub async fn contains_sedimentree_id(&self, doc_id: DocumentId) -> eyre::Result<bool> { | ||
| let (resp, rx) = futures::channel::oneshot::channel(); | ||
| self.cmd_tx | ||
| .send(Runtime2Cmd::CheckSedimentreeResident { doc_id, resp }) | ||
| .await | ||
| .map_err(|_| eyre::eyre!("task was found dead"))?; | ||
| rx.await | ||
| .map_err(|_| eyre::eyre!("caller dropped before response")) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Flatten the oneshot result; these methods do not type-check.
Both return expressions are missing ? after mapping the receiver error.
Proposed fix
rx.await
- .map_err(|_| eyre::eyre!("caller dropped before response"))
+ .map_err(|_| eyre::eyre!("caller dropped before response"))?Also applies to: 458-466
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_repo/runtime2/handle.rs` around lines 425 - 433, Update
contains_sedimentree_id and the corresponding method around the second reported
range to flatten the oneshot receiver result by applying ? after
rx.await.map_err(...). Preserve the existing error messages and return the
resolved boolean rather than a nested Result.
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/daybook_core/sync.rs (1)
244-253: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winRoute
BIG_SYNC_RPC_ALPNthroughprotocol_handler()in sync.rs.
local_sender()sends big-sync RPCs ontorpc_rx, whereBigSyncRpcWorker::handle_rpc_message(..., None)rejectsSubPartsrequests as unauthenticated.BIG_SYNC_RPC_ALPNis already handled this way inbootstrap.rs, so usebig_sync_rpc.protocol_handler()here as well.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/daybook_core/sync.rs` around lines 244 - 253, Update the BIG_SYNC_RPC_ALPN registration in the sync setup to use big_sync_rpc.protocol_handler() instead of IrohProtocol::with_sender(big_sync_rpc.local_sender()). Keep the existing big_repo RPC registration unchanged and ensure BigSync SubParts requests follow the authenticated protocol-handler path.src/big_repo/runtime2/io.rs (1)
221-233: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winDoc comment for
try_causal_decryptis attached to the wrong method.The "Causal decrypt: …" paragraph documents
try_causal_decrypt(Line 235) but is glued ontopersist_cgka_update_op, whose own doc follows immediately. Move it down to the method it describes.📝 Proposed fix
- /// Causal decrypt: decrypt `locator` + any ancestors whose keys are now - /// reachable. Mirrors - /// [`Document::try_causal_decrypt_content`](keyhive_core::principal::document::Document::try_causal_decrypt_content). - /// The returned [`CausalDecryptResult::complete`] includes the entrypoint - /// plus any ancestors decrypted along the causal chain. /// Persist a CGKA update operation emitted during encryption. /// /// Mirrors `keyhive_storage::persist_cgka_update_op`. Extracted as a /// `DocIo` method so the doc-worker doesn't hold concrete keyhive storage. fn persist_cgka_update_op( &self, op: keyhive_crypto::signed::Signed<beekem::operation::CgkaOperation>, ) -> F::Future<'_, eyre::Result<()>>; + /// Causal decrypt: decrypt `locator` + any ancestors whose keys are now + /// reachable. Mirrors + /// [`Document::try_causal_decrypt_content`](keyhive_core::principal::document::Document::try_causal_decrypt_content). + /// The returned [`CausalDecryptResult::complete`] includes the entrypoint + /// plus any ancestors decrypted along the causal chain. fn try_causal_decrypt(🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/io.rs` around lines 221 - 233, Move the causal-decrypt documentation block from above persist_cgka_update_op to directly above the try_causal_decrypt method, preserving its text and references. Keep the persist_cgka_update_op documentation attached only to that method.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/big_sync_core/lib.rs`:
- Around line 1847-1852: Update the test documentation near the waiter cleanup
scenario to reflect that SyncStatMachine::remove_peer now implements cleanup and
the test passes. Remove references to commented-out code, the obsolete FIXME,
and the test being red; describe the expected satisfied/removed waiter behavior
when its last peer is removed.
In `@src/big_sync/part_store/memory.rs`:
- Around line 1054-1081: Add an object_replayed guard to the in-memory replay
loop around the SubscriptionTarget::Object branch, allowing the object payload
to be appended only once per replay pass and preventing endless re-emission.
Reset object_replayed to false on both finalization-retry paths: when
begin_finalization() fails and when become_ready() fails, so concurrent
mutations replay the latest payload.
In `@src/big_sync/part_store/sqlite.rs`:
- Around line 240-254: The persisted access_level encoding and decoding must use
an explicit canonical mapping instead of Access enum discriminants. Update the
encode site and the decode match around doc_members to map Relay, Read, Edit,
and Admin directly to stable values 0–3, reject unknown values consistently, and
add a focused unit test covering round-trip mappings.
In `@src/daybook_core/sync.rs`:
- Around line 340-364: Fix per-peer cancellation in start_* implementations: in
src/daybook_core/sync.rs lines 340-364, create the map’s cancel token as a child
of self.keyhive_rpc_cancel and assign stream_cancel from cancel.clone(); in
src/big_repo/test.rs lines 3051-3111, create the map cancel token from
self.keyhive_rpc_cancel, remove the shadowing rebind, and have the spawned task
select on cancel.clone().
In `@src/daybook_core/sync/bootstrap.rs`:
- Line 281: Update the REPO_SYNC_ALPN registration in the bootstrap setup so
BigRepoRpcProtocolHandler::accept is reachable only through the authenticated
subduction_iroh transport. Reject or avoid routing connections established
through any unauthenticated endpoint before RPC messages are read, while
preserving authenticated Subduction endpoint handling.
---
Outside diff comments:
In `@src/big_repo/runtime2/io.rs`:
- Around line 221-233: Move the causal-decrypt documentation block from above
persist_cgka_update_op to directly above the try_causal_decrypt method,
preserving its text and references. Keep the persist_cgka_update_op
documentation attached only to that method.
In `@src/daybook_core/sync.rs`:
- Around line 244-253: Update the BIG_SYNC_RPC_ALPN registration in the sync
setup to use big_sync_rpc.protocol_handler() instead of
IrohProtocol::with_sender(big_sync_rpc.local_sender()). Keep the existing
big_repo RPC registration unchanged and ensure BigSync SubParts requests follow
the authenticated protocol-handler path.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: bc5b085e-7afe-46c4-af81-135550ca4edc
⛔ Files ignored due to path filters (1)
flake.lockis excluded by!**/*.lockand included by**/*
📒 Files selected for processing (46)
AGENTS.mdsrc/big_repo/backend.rssrc/big_repo/changes.rssrc/big_repo/ephemeral.rssrc/big_repo/keyhive.rssrc/big_repo/keyhive_storage.rssrc/big_repo/lib.rssrc/big_repo/rpc.rssrc/big_repo/runtime.rssrc/big_repo/runtime2/doc_worker.rssrc/big_repo/runtime2/group_part_worker.rssrc/big_repo/runtime2/hub.rssrc/big_repo/runtime2/io.rssrc/big_repo/runtime2/messages.rssrc/big_repo/runtime2/mod.rssrc/big_repo/runtime2/native.rssrc/big_repo/runtime2/test_support.rssrc/big_repo/sqlite_big_repo_store.rssrc/big_repo/test.rssrc/big_repo/test2/access_matrix.rssrc/big_repo/test2/capability.rssrc/big_repo/test2/cgka.rssrc/big_repo/test2/edge.rssrc/big_repo/test2/ephemeral.rssrc/big_repo/test2/harness/fixtures.rssrc/big_repo/test2/harness/heads.rssrc/big_repo/test2/harness/topo.rssrc/big_repo/test2/mod.rssrc/big_repo/test2/notifications.rssrc/big_repo/test2/topologies.rssrc/big_sync/lib.rssrc/big_sync/part_store.rssrc/big_sync/part_store/memory.rssrc/big_sync/part_store/sqlite.rssrc/big_sync/rpc.rssrc/big_sync/test.rssrc/big_sync/worker.rssrc/big_sync_core/bucket.rssrc/big_sync_core/cursor.rssrc/big_sync_core/lib.rssrc/big_sync_core/mpsc.rssrc/big_sync_core/rpc.rssrc/big_sync_core/tasks/peer_replay.rssrc/daybook_core/rt/switch.rssrc/daybook_core/sync.rssrc/daybook_core/sync/bootstrap.rs
💤 Files with no reviewable changes (2)
- src/big_repo/test2/cgka.rs
- src/big_repo/ephemeral.rs
🚧 Files skipped from review as they are similar to previous changes (13)
- src/big_repo/test2/mod.rs
- src/big_repo/runtime2/test_support.rs
- src/big_repo/test2/harness/heads.rs
- src/big_repo/test2/edge.rs
- src/big_repo/test2/access_matrix.rs
- src/big_repo/backend.rs
- src/big_repo/runtime2/mod.rs
- src/big_repo/test2/harness/fixtures.rs
- src/big_repo/runtime2/messages.rs
- src/big_repo/test2/harness/topo.rs
- src/big_repo/lib.rs
- src/big_repo/test2/capability.rs
- src/big_sync/lib.rs
| let access: u8 = row | ||
| .try_get::<i64, _>("access_level")? | ||
| .try_into() | ||
| .expect(ERROR_IMPOSSIBLE); | ||
| let access = match access { | ||
| 0 => keyhive_core::access::Access::Relay, | ||
| 1 => keyhive_core::access::Access::Read, | ||
| 2 => keyhive_core::access::Access::Edit, | ||
| 3 => keyhive_core::access::Access::Admin, | ||
| other => panic!("invalid persisted access_level {other}"), | ||
| }; | ||
| doc_members | ||
| .entry(obj_id) | ||
| .or_default() | ||
| .insert(principal, access); |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
keyhive_core access Access enum variant discriminant order Relay Read Edit Admin
💡 Result:
In the keyhive_core crate, the Access enum defines access levels for capabilities, which are ordered from least to most permissive [1]. As of the current implementation, the variants of the Access enum in order are Relay, Read, Edit, and Admin [1]. The order reflects the hierarchy where later levels imply all earlier levels [1]. The enum variants are: 1. Relay: The least permissive level; a node with only Relay access does not have decryption (Read) access to the underlying data [1]. 2. Read: The ability to read (decrypt) the content of a document [1]. 3. Edit: The ability to edit (append operations to) the content of a document [1]. 4. Admin: The ability to revoke members of a group [1]. This order was updated in commit 179c7c5, which renamed the previous Pull and Write variants to Relay and Edit, respectively [1].
Citations:
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Files likely defining Access:\n'
fd -a 'lib.rs|access.rs|.*access.*' . | head -80
printf '\nSearch Access enum definitions/usages:\n'
rg -n "enum Access|struct Access|Access::|access_level|set_doc_members|add_doc_member|reconcile_group_part_batch|try_get::<i64" .
printf '\nSqlite part_store lines around access rehydration:\n'
fd -a 'sqlite.rs' src | while read -r f; do
echo "===== $f ====="
wc -l "$f"
sed -n '220,265p;900,920p;1448,1462p' "$ f" 2>/dev/null || true
doneRepository: dman-os/townframe
Length of output: 40573
🏁 Script executed:
#!/bin/bash
set -euo pipefail
cat -n src/big_sync/part_store/sqlite.rs | sed -n '1740,1810p'
printf '\nkeyhive dependency declarations:\n'
rg -n 'keyhive_core|keyhive' Cargo.toml Cargo.lock pyproject.toml 2>/dev/null | head -120 || true
printf '\nRepo config for running/checking tests:\n'
for f in Cargo.toml Cargo.lock pyproject.toml; do
[ -f "$f" ] && wc -l "$f"
doneRepository: dman-os/townframe
Length of output: 3932
Keep the persisted access_level mapping explicit by encoding the access enum.
The current as u8 encode at binding = i64::from(*access as u8) and the literal 0..3 decode here couple authorization after restart to variant discriminants. Use a small canonical mapping for both encode and decode, ideally covered by a unit test, rather than relying on the enum order.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_sync/part_store/sqlite.rs` around lines 240 - 254, The persisted
access_level encoding and decoding must use an explicit canonical mapping
instead of Access enum discriminants. Update the encode site and the decode
match around doc_members to map Relay, Read, Edit, and Admin directly to stable
values 0–3, reject unknown values consistently, and add a focused unit test
covering round-trip mappings.
| let cancel = CancellationToken::new(); | ||
| if let Some(previous) = self | ||
| .keyhive_rpc_cancels | ||
| .lock() | ||
| .await | ||
| .insert(peer_id, cancel.clone()) | ||
| { | ||
| previous.cancel(); | ||
| } | ||
|
|
||
| let client = | ||
| big_repo::rpc::IrohBigRepoRpcClient::new(self.router.endpoint().clone(), endpoint_addr); | ||
| let mut changes = | ||
| tokio::time::timeout(Duration::from_secs(5), client.subscribe_keyhive_changes(64)) | ||
| .await | ||
| .map_err(|_| ferr!("timed out installing Keyhive RPC subscription"))??; | ||
| let ready = tokio::time::timeout(Duration::from_secs(5), changes.recv()) | ||
| .await | ||
| .map_err(|_| ferr!("timed out waiting for Keyhive RPC subscription readiness"))?? | ||
| .ok_or_else(|| ferr!("Keyhive RPC stream closed before readiness"))?; | ||
| if !ready.initial { | ||
| eyre::bail!("Keyhive RPC stream did not send its readiness event"); | ||
| } | ||
|
|
||
| let stream_cancel = self.keyhive_rpc_cancel.child_token(); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Per-peer Keyhive-RPC cancellation is wired to the wrong token in both implementations. In each start_* the token placed in keyhive_rpc_cancels is not the token the spawned subscription task selects on (the task uses a fresh child of the global keyhive_rpc_cancel), so the per-peer stop_* cancels a token nothing awaits — per-peer teardown becomes a no-op and reconnects can leak lingering tasks.
src/daybook_core/sync.rs#L340-L364: makecancela child ofself.keyhive_rpc_canceland setstream_cancel = cancel.clone()so the task observes the per-peer token.src/big_repo/test.rs#L3051-L3111: make the mapcancel(Line 3053) a child ofself.keyhive_rpc_cancel, drop the shadowing rebind at Line 3077, and have the task select oncancel.clone().
📍 Affects 2 files
src/daybook_core/sync.rs#L340-L364(this comment)src/big_repo/test.rs#L3051-L3111
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/daybook_core/sync.rs` around lines 340 - 364, Fix per-peer cancellation
in start_* implementations: in src/daybook_core/sync.rs lines 340-364, create
the map’s cancel token as a child of self.keyhive_rpc_cancel and assign
stream_cancel from cancel.clone(); in src/big_repo/test.rs lines 3051-3111,
create the map cancel token from self.keyhive_rpc_cancel, remove the shadowing
rebind, and have the spawned task select on cancel.clone().
| peer_key_fn: Arc::new(|endpoint_id| PeerId::new(*endpoint_id.as_bytes())), | ||
| }, | ||
| ) | ||
| .accept(big_repo::rpc::REPO_SYNC_ALPN, repo_rpc.protocol_handler()) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm auth is enforced inside the repo RPC protocol handler and no callers still rely on the removed wrapper.
rg -nP --type=rust -C3 'fn protocol_handler' src/big_repo/rpc.rs
rg -nP --type=rust -C3 'AuthenticatedIrohProtocol|peer_key_fn|authenticated' src/big_repo/rpc.rs
rg -nP --type=rust -C2 'REPO_SYNC_ALPN'Repository: dman-os/townframe
Length of output: 1121
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== rust files matching bootstrap.rs / big_repo rpc =="
git ls-files | rg '(^|/)bootstrap\.rs$|(^|/)rpc\.rs$|(^|/)protocol'
echo
echo "== bootstrap.rs relevant lines =="
if [ -f src/daybook_core/sync/bootstrap.rs ]; then
sed -n '240,325p' src/daybook_core/sync/bootstrap.rs | cat -n
fi
echo
echo "== big_repo rpc outline =="
ast-grep outline src/big_repo/rpc.rs --view expanded || true
echo
echo "== rpc.rs protocol handler and auth-related content =="
sed -n '70,145p' src/big_repo/rpc.rs | cat -n
rg -n --type=rust -C4 'protocol_handler|AuthenticatedIrohProtocol|peer_key_fn|authenticated|authentication|authorize|authorize_endpoint|Endpoint|accept|REPO_SYNC_ALPN' .
echo
echo "== auth imports/usages =="
rg -n --type=rust 'AuthenticatedIrohProtocol|peer_key_fn|peer_key|PeerKey|EndpointAuth|Authorizer|authorize|authorize_endpoint|REPO_SYNC_ALPN|BigRepoRpcProtocolHandler' src || true
echo
echo "== BigRepoRpcProtocolHandler implementation =="
rg -n --type=rust -C8 'BigRepoRpcProtocolHandler|type .*Protocol|impl .*Protocol|fn handle|rpc_tx|peer_map' src/big_repo src/daybook_core/sync || trueRepository: dman-os/townframe
Length of output: 50374
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== candidate file sizes =="
wc -l src/big_repo/rpc.rs src/big_repo/start.rs src/big_repo/runtime.rs src/daybook_core/sync/bootstrap.rs src/big_repo/test.rs 2>/dev/null || true
echo
echo "== src/big_repo/rpc.rs top =="
sed -n '1,180p' src/big_repo/rpc.rs | cat -n
echo
echo "== REPO_SYNC_ALPN definition/usages =="
rg -n --type=rust -C8 'REPO_SYNC_ALPN' src
echo
echo "== rpc handler type trait usages =="
rg -n --type=rust -C8 'impl .*ProtocolHandler|trait ProtocolHandler|type .*ProtocolHandler|protocol_handler|AcceptError|BigRepoRpcProtocolHandler' src/big_repo/rpc.rs src -g'*.rs' | head -n 400
echo
echo "== bootstrap.rs router/bootstrap context =="
sed -n '1,80p' src/daybook_core/sync/bootstrap.rs | cat -n
sed -n '230,330p' src/daybook_core/sync/bootstrap.rs | cat -nRepository: dman-os/townframe
Length of output: 50008
Keep REPO_SYNC_ALPN routed through authenticated Subduction endpoints.
BigRepoRpcProtocolHandler::accept only reads conn.remote_id() and looks it up in peer_map, without validating the peer against the authenticated Subduction transport. Reject connections that are not established via the authenticated subduction_iroh path before reading RPC messages, otherwise this handler opens an unauthenticated repo-RPC channel.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/daybook_core/sync/bootstrap.rs` at line 281, Update the REPO_SYNC_ALPN
registration in the bootstrap setup so BigRepoRpcProtocolHandler::accept is
reachable only through the authenticated subduction_iroh transport. Reject or
avoid routing connections established through any unauthenticated endpoint
before RPC messages are read, while preserving authenticated Subduction endpoint
handling.
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (7)
src/big_repo/test.rs (1)
2971-3037: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
stop_keyhive_rpccancels a token nothing listens to.Line 2973 creates the per-peer
canceland stores it inkeyhive_rpc_cancels, but line 2997 shadows that binding withself.keyhive_rpc_cancel.child_token(), and the spawned task selects on the shadowed child token. So neitherstop_keyhive_rpc(line 3035) nor theprevious.cancel()at line 2980 has any effect — the subscription task keeps running afterdisconnect_fromand keeps callingrepo.sync_keyhive_with_peer(peer_id, …)against a peer the test believes is disconnected. Only the globalkeyhive_rpc_cancelat shutdown stops it.That silently defeats disconnection isolation in tests such as
run_sync_backend_case, which relies ondisconnect_from/stop_big_sync_withto stop background propagation.🐛 Proposed fix: make the per-peer token a child and use it in the task
async fn start_keyhive_rpc(&self, remote: &SyncRepoNode) -> Res<()> { let peer_id = remote.peer_id(); - let cancel = CancellationToken::new(); + let cancel = self.keyhive_rpc_cancel.child_token(); if let Some(previous) = self .keyhive_rpc_cancels .lock() .await .insert(peer_id, cancel.clone()) { previous.cancel(); } @@ let repo = Arc::clone(&self.repo); - let cancel = self.keyhive_rpc_cancel.child_token(); self.keyhive_rpc_tasks🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test.rs` around lines 2971 - 3037, Use the per-peer cancellation token created in start_keyhive_rpc for the spawned subscription task: make that token a child of keyhive_rpc_cancel if needed, but remove the shadowing assignment from self.keyhive_rpc_cancel.child_token(). Ensure the task’s select listens to the same token stored in keyhive_rpc_cancels so stop_keyhive_rpc and previous.cancel() terminate the corresponding subscription.src/big_repo/runtime2/native.rs (2)
343-412: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
sedimentree_headsdiscards correctly computed heads and returns an empty frontier.When
empty_cached_countsisSome(cache reported an empty frontier) but durable storage actually has commits/fragments, the function reaches line 392-410, computes the realheadsfrom durable storage, logs a diagnostic warning, and then unconditionally returnsOk(Vec::new())— discarding the just-computedheadsinstead of returning them. This means any time the in-memory cache is stale/empty while durable storage is populated, callers (includingquery_head_stateinsrc/big_repo/runtime2/doc_worker.rs, and tests liketier7_local_mutation_quiescence_keyhive_state_and_notificationinsrc/big_repo/test2/notifications.rs) will observe an empty sedimentree frontier even though real head data exists.🐛 Proposed fix
let heads = sedimentree_heads_payload(&tree); if let Some((cached_loose, cached_fragments)) = empty_cached_counts { tracing::warn!( ?sed_id, cached_loose, cached_fragments, durable_loose = loose_commits.len(), durable_fragments = fragments.len(), durable_heads = heads.len(), "cached Sedimentree reported empty heads; compared durable state" ); - return Ok(Vec::new()); } Ok(heads.iter().map(|h| CommitId::new(h.0)).collect())🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/native.rs` around lines 343 - 412, Update sedimentree_heads so that when empty_cached_counts is present but durable storage contains commits or fragments, it returns the computed heads from sedimentree_heads_payload(&tree) instead of Ok(Vec::new()). Preserve the diagnostic warning and the existing empty result when both cached and durable state are empty.
1433-1464: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPeriodic keyhive maintenance tasks silently swallow failures instead of crashing at the task boundary.
Both the cache-refresh and archive-compaction loops propagate errors via
?from an infiniteloop, rather than calling.unwrap()/.expect()as the siblinggroup_part_worker(line 1408) andlistener(line 1415) tasks in this same function do. These are unexpected failures that are not reported through a channel, so per the guideline they should crash visibly rather than return anErrthat may be silently absorbed by whatever consumeschild_tasks' results — leaving keyhive cache refresh/archive compaction stopped without any visible signal.🔧 Proposed fix
loop { timer.sleep(std::time::Duration::from_secs(2)).await; - kh_proto - .refresh_cache() - .await - .map_err(|error| ferr!("keyhive cache refresh failed: {error}"))?; + kh_proto + .refresh_cache() + .await + .expect("keyhive cache refresh failed"); }loop { timer.sleep(std::time::Duration::from_secs(300)).await; - kh_proto - .compact(keyhive_archive_id) - .await - .map_err(|error| ferr!("keyhive archive compaction failed: {error}"))?; + kh_proto + .compact(keyhive_archive_id) + .await + .expect("keyhive archive compaction failed"); }As per coding guidelines, "For unexpected failures at the final Tokio task boundary, call
.unwrap()unless the task is explicitly expected to fail and reports failure through a channel."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/native.rs` around lines 1433 - 1464, Update the periodic maintenance tasks spawned in the cache-refresh and archive-compaction blocks to unwrap the futures at the final task boundary, rather than propagating errors with ?. Preserve the existing error messages from refresh_cache and compact, and make both child tasks crash visibly on unexpected failures like the sibling group_part_worker and listener tasks.Source: Coding guidelines
src/big_repo/keyhive_storage.rs (1)
350-401: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winMake
BigRepoKeyhiveStorage::FsuseFsKeyhiveStoragefor events, or remove the dead hard-link path.
BigRepoKeyhiveStorage::fsconstructsSelf::Fs { events, archives }, buteventsis alwaysSqliteBigRepoStoreand allBigRepoKeyhiveStorageevent dispatch still goes throughevents.save_keyhive_event/load_keyhive_events/delete_keyhive_event. The hard-linksave_file_if_absentpath onFsKeyhiveStorage::save_eventis only reached ifFsKeyhiveStorageis used standalone as aKeyhiveStorageorFs.eventsis changed to that type.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/keyhive_storage.rs` around lines 350 - 401, Update BigRepoKeyhiveStorage::fs and the Fs variant so its events field uses FsKeyhiveStorage instead of SqliteBigRepoStore, ensuring save_event, load_events, and delete_event dispatch through the filesystem-backed implementation. Preserve the existing Memory and MemoryLegacy paths, and ensure the FsKeyhiveStorage hard-link behavior is reachable for Fs instances.src/big_sync/part_store/memory.rs (2)
1609-1631: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winThese "filter" assertions don't assert anything.
The non-reader half of
syncability_filter_drops_non_readable_events(and the revocation half ofsyncability_filter_updates, lines 1677-1688) only confirms the mutation call returnedOk; the filtering behavior is never observed. A shorttokio::time::timeout(..., rx2.recv())that expects an elapse — assqlite_membership_cache_rehydrates_after_restartdoes insrc/big_sync/part_store/sqlite.rslines 2365-2373 — would actually cover the claim.As per coding guidelines, "Proactively improve weak tests: flag useless or shallow tests for removal".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync/part_store/memory.rs` around lines 1609 - 1631, The non-reader branch of syncability_filter_drops_non_readable_events only verifies that add_obj_to_parts succeeds and never observes rx2. Replace the drop-only check with a short tokio::time::timeout around rx2.recv(), asserting the timeout elapses without receiving an event, and apply the same no-event assertion to the revocation branch of syncability_filter_updates.Source: Coding guidelines
1130-1154: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
object_replay_pendingis never re-armed, so object mutations racing the replay→live handoff are lost.The flag is cleared unconditionally at line 1130 and never set back to
true. Meanwhileset_obj_payload's no-parts branch (lines 666-678) drops theObjectChangedwhenevermark_dirty()reports the subscription is still replaying, on the assumption that replay will pick the payload up. It won't: subsequent loop iterations skip the object block entirely.The SQLite twin re-arms on every retry path (
src/big_sync/part_store/sqlite.rslines 1813, 1824, 1828); mirror that here.🐛 Proposed fix re-arming object replay on the retry paths
if !marker_sent { if !pending_for_replay.begin_finalization() { + object_replay_pending = true; continue; } if tx.send(SubEvent::ReplayComplete).await.is_err() { return; } marker_sent = true; if pending_for_replay.become_ready() { return; } + object_replay_pending = true; } else if pending_for_replay.become_ready() { return; + } else { + object_replay_pending = true; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync/part_store/memory.rs` around lines 1130 - 1154, Re-arm object replay whenever the replay-to-live loop retries after clearing object_replay_pending, matching the SQLite implementation. Update the retry paths around pending_for_replay.begin_finalization(), pending_for_replay.become_ready(), and any continue caused by replay activity so object_replay_pending is set true before continuing, ensuring racing object mutations are replayed.src/big_sync_core/lib.rs (1)
601-616: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
WaitForFullSynccan panic if a referenced peer was removed (or never set).Every other handler here (
handle_set_peer_strat,handle_sync_completed,handle_decide_peer_strat_err, etc.) guards against a peer being absent fromself.peersvialet Some(peer_state) = self.peers.get_mut(&peer_id) else { assert!(self.all_seen_peer.contains(&peer_id), "fishy"); return; }, tolerating the race where a peer was previously set and later removed. ThisWaitForFullSyncarm skips that guard entirely and callsrefresh_peer_replay_worker(*peer_id, true)directly, which doesself.peers.get_mut(&peer_id).expect(ERROR_UNRECONIZED)(Line 1094-1096) and will panic for anypeer_idinpeer_idsthat isn't currently registered — crashing the whole sync actor instead of degrading gracefully like the rest of this file.🛠️ Proposed fix
BigSyncEvent::WaitForFullSync(evt) => { let WaitForFullSyncEvent { waiter_id, peer_ids, part_ids, } = evt; for peer_id in &peer_ids { - self.refresh_peer_replay_worker(*peer_id, true); + if self.peers.contains_key(peer_id) { + self.refresh_peer_replay_worker(*peer_id, true); + } else { + assert!(self.all_seen_peer.contains(peer_id), "fishy"); + } } self.stat_machine .add_full_sync_waiter(waiter_id, peer_ids, part_ids); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync_core/lib.rs` around lines 601 - 616, Update the WaitForFullSync arm in handle_evt to validate each peer_id against self.peers before calling refresh_peer_replay_worker, tolerating absent peers with the same all_seen_peer assertion-and-return pattern used by the other handlers. Ensure missing or removed peers do not reach refresh_peer_replay_worker and panic, while registered peers continue through the existing full-sync waiter flow.
🧹 Nitpick comments (10)
src/big_repo/test.rs (2)
3128-3136: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winCancel the Keyhive RPC tasks before stopping the repo.
The spawned subscription tasks hold
Arc<BigRepo>and callsync_keyhive_with_peer, yetself.stop_token.stop()runs beforekeyhive_rpc_cancel.cancel(). In-flight calls then race against an already-stopped runtime; the failure is swallowed atdebug!, but the ordering is avoidable noise.♻️ Suggested ordering
self.endpoint.close().await; + self.keyhive_rpc_cancel.cancel(); + self.keyhive_rpc_tasks + .stop(Duration::from_secs(5)) + .await + .wrap_err("failed stopping test Keyhive RPC subscriptions")?; self.stop_token.stop().await?; self.big_sync_stop.stop().await?; - self.keyhive_rpc_cancel.cancel(); - self.keyhive_rpc_tasks - .stop(Duration::from_secs(5)) - .await - .wrap_err("failed stopping test Keyhive RPC subscriptions")?; self.repo_rpc_stop.stop().await?;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test.rs` around lines 3128 - 3136, Reorder shutdown in the test teardown so keyhive RPC subscriptions are cancelled before invoking self.stop_token.stop(). Keep the existing keyhive_rpc_tasks.stop cleanup and remaining shutdown calls intact, ensuring sync_keyhive_with_peer tasks stop before the repository runtime is stopped.
3784-3793: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDowngrading
ChangedObjecttoNoopmakes the assertion unfalsifiable.
client.stop_big_sync_with(&server)at line 3755 already runs before both mutations, so the stated justification ("a prior subscription may deliver the remote mutation") does not obviously apply to this ordering. As written, any case where the payloads happen to match silently accepts a no-op, so a genuine regression where the backend stops performing work would still pass.run_sync_backend_put_doc_conflict_case(lines 3936-3971) shows the deterministic alternative: isolate the race, then assert the exact outcome.Please either document the concrete remaining race window, or make the pre-state deterministic and keep the strict expectation.
As per coding guidelines: "Do not use hacks or shallow fixes merely to make tests pass; implement correct solutions" and "Proactively improve weak tests".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test.rs` around lines 3784 - 3793, Remove the conditional downgrade from the test’s expected outcome: keep ChangedObject strict when that is the scenario under test, rather than converting it to Noop based on matching payloads. Make the pre-state deterministic by isolating any subscription race, following the approach in run_sync_backend_put_doc_conflict_case, and retain the exact completion assertion.Source: Coding guidelines
src/big_sync/stress_support.rs (1)
367-400: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDuplicate journal entry in the no-mutator fallback.
Line 442 records
create(no-mutator) node=…and line 458 records the same prefix again with the value appended. Drop the firstjournal.record, it only adds noise to failure dumps.Also applies to: 438-464
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync/stress_support.rs` around lines 367 - 400, Remove the earlier duplicate journal.record call in the no-mutator fallback, leaving only the later create(no-mutator) entry that includes the value. Keep the mutation behavior and the remaining journal entry unchanged.src/big_repo/runtime2/native.rs (1)
728-745: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winO(n²) dedup scan for causal ancestors.
Each entry of
state.completeis checked againstcompletewith a linear.any()scan, so the loop is quadratic in the number of causal ancestors. For documents with large causal closures this could become a meaningful cost on the decrypt hot path.Consider tracking seen
content_refs in aHashSet/BTreeSetinstead, if the ref type supports the required bound.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/native.rs` around lines 728 - 745, Replace the linear deduplication scan in the loop over state.complete with a set of seen content_ref values, initializing it with the entrypoint reference and updating it as each unique reference is added to complete. Use HashSet or BTreeSet according to the content_ref trait bounds, while preserving the existing deserialization fallback behavior.src/big_sync/part_store/memory.rs (1)
419-435: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDuplicated unlink logic with
flush().Lines 391-407 in
flush()repeat exactly whatremove_subscriptiondoes. Collect the drained ids into a localVecfirst, then callself.bus.remove_subscription(sub_id)to keep one implementation.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync/part_store/memory.rs` around lines 419 - 435, Refactor flush() to collect the drained subscription IDs into a local Vec, then invoke remove_subscription for each ID instead of duplicating the unlink logic. Reuse the existing remove_subscription method to remove entries from all related subscription maps while preserving flush() behavior.src/big_repo/test2/stress.rs (2)
247-264: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
tempdir()?.keep()leaks a directory per booted node, forever.
keep()defuses the cleanup guard, which is necessary so the path survivesrestart_node, but nothing ever removes it — each stress run permanently accumulates node databases under the temp dir. Sincenode_pathsalready tracks every path, a teardown step (or aTempDirowned by the fixture and dropped at the end) would reclaim them.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/stress.rs` around lines 247 - 264, Update boot_node so directories recorded in node_paths are reclaimed during fixture teardown instead of permanently leaked by tempdir().keep(). Preserve the path lifetime needed by restart_node, and add cleanup for every tracked node path at the end of the stress fixture lifecycle.
147-197: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winCheck-then-set on
shared_edit_group_idis not atomic.The
Nonecheck at line 158 releases the lock before line 169 writes it, so two concurrentedit_groupcallers can each create a group and the second overwrites the first — leaving the two nodes on different edit groups and producing a confusing convergence failure rather than a clear error. Holding one guard across both the read and the write (or usingget_or_inserton a single lock scope) removes the window.Note
prepare_clusternormally pre-seeds the id, so this only bites on paths that reachedit_groupbefore/without bootstrap.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/test2/stress.rs` around lines 147 - 197, The shared_edit_group_id check and initialization in edit_group are not atomic, allowing concurrent callers to create different groups. Hold a single lock guard across reading the existing ID and, when absent, creating the group and storing its ID; reuse that guard through the initialization path while preserving the existing persisted-group and membership behavior.src/big_sync/part_store/sqlite.rs (3)
1709-1713: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winReplay reads the entire part history per iteration.
u32::MAXas the limit means every loop pass materializes all events for all subscribed parts, and the loop re-runs on eachcontinue(new events, or a failedbegin_finalization/become_ready). The in-memory twin caps a pass at 50 events (src/big_sync/part_store/memory.rs line 1101). A bounded page size here would keep replay memory and latency proportional to the batch rather than to total history.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync/part_store/sqlite.rs` around lines 1709 - 1713, Update the replay loop’s list_events call to use the bounded page size established by the in-memory implementation instead of u32::MAX. Preserve the existing cursor handling and continuation behavior so each iteration processes at most 50 events across the subscribed parts.
747-757: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAdd an
(scope_id, obj_id)index onbig_sync_pending_members.Both hot lookups (
set_obj_payloadline 909-912 andobj_partsline 1010-1012) filter byscope_id, obj_id, which the(scope_id, part_id, obj_id)primary key cannot serve — mirroringbig_sync_members_obj_idx(lines 741-746) avoids a full-table scan as latent memberships accumulate.🗂️ Proposed index
.execute(&mut *tx) .await?; + sqlx::query( + "CREATE INDEX IF NOT EXISTS big_sync_pending_members_obj_idx + ON big_sync_pending_members(scope_id, obj_id)", + ) + .execute(&mut *tx) + .await?;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync/part_store/sqlite.rs` around lines 747 - 757, Add a non-unique index on `(scope_id, obj_id)` when creating `big_sync_pending_members`, alongside the existing table definition and primary key. Name and define it consistently with the analogous `big_sync_members_obj_idx` index so the `set_obj_payload` and `obj_parts` lookups can use it.
495-500: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse
Option<SubEvent>instead of aReplayCompletesentinel.
SubEvent::ReplayCompleteis overloaded here to mean "this event has no object-level projection", requiring the!matches!(...)guard at line 522.MemoryPartStoreScopeState::flushmodels the same thing asOption<SubEvent>(src/big_sync/part_store/memory.rs lines 293-307); matching that removes the sentinel and the risk of a realReplayCompleteleaking into the projection.Also applies to: 522-530
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_sync/part_store/sqlite.rs` around lines 495 - 500, Update the event projection logic around the match handling SubEvent::Removed, SubEvent::ObjectChanged, and SubEvent::ReplayComplete to return an Option<SubEvent>, using None for events without an object-level projection instead of SubEvent::ReplayComplete as a sentinel. Align the downstream projection handling and remove the !matches! guard so only Some events are emitted, while preserving the existing IDs and cloned event behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/big_sync/part_store/memory.rs`:
- Around line 660-665: Update the object-notification permission calculation in
the `permitted` chain to default to true when `doc_members` has no entry,
matching `SqlitePartStore::publish`. Preserve the existing `is_fetcher()` result
when a member entry exists, so only the missing-membership case changes.
In `@src/big_sync/part_store/sqlite.rs`:
- Around line 931-946: Keep pending membership materialization atomic with the
transaction that reads pending_part_ids in the live_part_ids.is_empty() branch.
Update add_obj_to_parts or this branch so materialization uses the existing
transaction, or re-reads and validates pending rows within add_obj_to_parts’
transaction before inserting; preserve the current publish behavior when no
pending memberships remain.
- Around line 1605-1607: Update the Added-event payload construction in
list_events to handle nullable added_payload_json without panicking: use
added_payload when present, otherwise fall back to the joined objs.payload_json
value. Preserve the existing expectation only where neither source is available,
and ensure stale NULL rows load gracefully.
In `@src/big_sync/stress_support.rs`:
- Around line 750-757: Update the final settle wait in the surrounding stress
fixture flow to pass the caller-provided settle_timeout instead of
Duration::from_secs(60). Preserve the existing final:post-connect label and
wait_for_settled behavior while ensuring larger caller budgets are honored.
- Around line 271-275: Validate node_count before the boot loop so values
exceeding the peer-seed range fail immediately instead of being cast to u8.
Update the loop around fixture.boot_node and assert_cluster_alignment to use the
validated range, preserving normal behavior for supported counts and rejecting
out-of-range counts explicitly.
---
Outside diff comments:
In `@src/big_repo/keyhive_storage.rs`:
- Around line 350-401: Update BigRepoKeyhiveStorage::fs and the Fs variant so
its events field uses FsKeyhiveStorage instead of SqliteBigRepoStore, ensuring
save_event, load_events, and delete_event dispatch through the filesystem-backed
implementation. Preserve the existing Memory and MemoryLegacy paths, and ensure
the FsKeyhiveStorage hard-link behavior is reachable for Fs instances.
In `@src/big_repo/runtime2/native.rs`:
- Around line 343-412: Update sedimentree_heads so that when empty_cached_counts
is present but durable storage contains commits or fragments, it returns the
computed heads from sedimentree_heads_payload(&tree) instead of Ok(Vec::new()).
Preserve the diagnostic warning and the existing empty result when both cached
and durable state are empty.
- Around line 1433-1464: Update the periodic maintenance tasks spawned in the
cache-refresh and archive-compaction blocks to unwrap the futures at the final
task boundary, rather than propagating errors with ?. Preserve the existing
error messages from refresh_cache and compact, and make both child tasks crash
visibly on unexpected failures like the sibling group_part_worker and listener
tasks.
In `@src/big_repo/test.rs`:
- Around line 2971-3037: Use the per-peer cancellation token created in
start_keyhive_rpc for the spawned subscription task: make that token a child of
keyhive_rpc_cancel if needed, but remove the shadowing assignment from
self.keyhive_rpc_cancel.child_token(). Ensure the task’s select listens to the
same token stored in keyhive_rpc_cancels so stop_keyhive_rpc and
previous.cancel() terminate the corresponding subscription.
In `@src/big_sync_core/lib.rs`:
- Around line 601-616: Update the WaitForFullSync arm in handle_evt to validate
each peer_id against self.peers before calling refresh_peer_replay_worker,
tolerating absent peers with the same all_seen_peer assertion-and-return pattern
used by the other handlers. Ensure missing or removed peers do not reach
refresh_peer_replay_worker and panic, while registered peers continue through
the existing full-sync waiter flow.
In `@src/big_sync/part_store/memory.rs`:
- Around line 1609-1631: The non-reader branch of
syncability_filter_drops_non_readable_events only verifies that add_obj_to_parts
succeeds and never observes rx2. Replace the drop-only check with a short
tokio::time::timeout around rx2.recv(), asserting the timeout elapses without
receiving an event, and apply the same no-event assertion to the revocation
branch of syncability_filter_updates.
- Around line 1130-1154: Re-arm object replay whenever the replay-to-live loop
retries after clearing object_replay_pending, matching the SQLite
implementation. Update the retry paths around
pending_for_replay.begin_finalization(), pending_for_replay.become_ready(), and
any continue caused by replay activity so object_replay_pending is set true
before continuing, ensuring racing object mutations are replayed.
---
Nitpick comments:
In `@src/big_repo/runtime2/native.rs`:
- Around line 728-745: Replace the linear deduplication scan in the loop over
state.complete with a set of seen content_ref values, initializing it with the
entrypoint reference and updating it as each unique reference is added to
complete. Use HashSet or BTreeSet according to the content_ref trait bounds,
while preserving the existing deserialization fallback behavior.
In `@src/big_repo/test.rs`:
- Around line 3128-3136: Reorder shutdown in the test teardown so keyhive RPC
subscriptions are cancelled before invoking self.stop_token.stop(). Keep the
existing keyhive_rpc_tasks.stop cleanup and remaining shutdown calls intact,
ensuring sync_keyhive_with_peer tasks stop before the repository runtime is
stopped.
- Around line 3784-3793: Remove the conditional downgrade from the test’s
expected outcome: keep ChangedObject strict when that is the scenario under
test, rather than converting it to Noop based on matching payloads. Make the
pre-state deterministic by isolating any subscription race, following the
approach in run_sync_backend_put_doc_conflict_case, and retain the exact
completion assertion.
In `@src/big_repo/test2/stress.rs`:
- Around line 247-264: Update boot_node so directories recorded in node_paths
are reclaimed during fixture teardown instead of permanently leaked by
tempdir().keep(). Preserve the path lifetime needed by restart_node, and add
cleanup for every tracked node path at the end of the stress fixture lifecycle.
- Around line 147-197: The shared_edit_group_id check and initialization in
edit_group are not atomic, allowing concurrent callers to create different
groups. Hold a single lock guard across reading the existing ID and, when
absent, creating the group and storing its ID; reuse that guard through the
initialization path while preserving the existing persisted-group and membership
behavior.
In `@src/big_sync/part_store/memory.rs`:
- Around line 419-435: Refactor flush() to collect the drained subscription IDs
into a local Vec, then invoke remove_subscription for each ID instead of
duplicating the unlink logic. Reuse the existing remove_subscription method to
remove entries from all related subscription maps while preserving flush()
behavior.
In `@src/big_sync/part_store/sqlite.rs`:
- Around line 1709-1713: Update the replay loop’s list_events call to use the
bounded page size established by the in-memory implementation instead of
u32::MAX. Preserve the existing cursor handling and continuation behavior so
each iteration processes at most 50 events across the subscribed parts.
- Around line 747-757: Add a non-unique index on `(scope_id, obj_id)` when
creating `big_sync_pending_members`, alongside the existing table definition and
primary key. Name and define it consistently with the analogous
`big_sync_members_obj_idx` index so the `set_obj_payload` and `obj_parts`
lookups can use it.
- Around line 495-500: Update the event projection logic around the match
handling SubEvent::Removed, SubEvent::ObjectChanged, and
SubEvent::ReplayComplete to return an Option<SubEvent>, using None for events
without an object-level projection instead of SubEvent::ReplayComplete as a
sentinel. Align the downstream projection handling and remove the !matches!
guard so only Some events are emitted, while preserving the existing IDs and
cloned event behavior.
In `@src/big_sync/stress_support.rs`:
- Around line 367-400: Remove the earlier duplicate journal.record call in the
no-mutator fallback, leaving only the later create(no-mutator) entry that
includes the value. Keep the mutation behavior and the remaining journal entry
unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 7f672bb9-6ec1-49d9-9a94-168aa0d86863
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lockand included by**/*
📒 Files selected for processing (39)
Cargo.tomlsrc/big_repo/keyhive.rssrc/big_repo/keyhive_storage.rssrc/big_repo/runtime.rssrc/big_repo/runtime2/doc_worker.rssrc/big_repo/runtime2/group_part_worker.rssrc/big_repo/runtime2/hub.rssrc/big_repo/runtime2/io.rssrc/big_repo/runtime2/messages.rssrc/big_repo/runtime2/mod.rssrc/big_repo/runtime2/native.rssrc/big_repo/sqlite_big_repo_store.rssrc/big_repo/test.rssrc/big_repo/test2/access_matrix.rssrc/big_repo/test2/convergence.rssrc/big_repo/test2/edge.rssrc/big_repo/test2/fragmentation.rssrc/big_repo/test2/harness/fixtures.rssrc/big_repo/test2/harness/topo.rssrc/big_repo/test2/mod.rssrc/big_repo/test2/notifications.rssrc/big_repo/test2/restart.rssrc/big_repo/test2/stress.rssrc/big_sync/backend.rssrc/big_sync/part_store.rssrc/big_sync/part_store/memory.rssrc/big_sync/part_store/sqlite.rssrc/big_sync/rpc.rssrc/big_sync/stress_support.rssrc/big_sync/test.rssrc/big_sync/test/stress.rssrc/big_sync_core/cursor.rssrc/big_sync_core/lib.rssrc/big_sync_core/rpc.rssrc/big_sync_core/tasks/decide_peer_strat.rssrc/big_sync_core/tasks/peer_replay.rssrc/daybook_core/blobs/sync.rssrc/xtask/Cargo.tomlsrc/xtask/keyhive_demo.rs
💤 Files with no reviewable changes (2)
- src/daybook_core/blobs/sync.rs
- src/big_sync/backend.rs
🚧 Files skipped from review as they are similar to previous changes (20)
- src/big_repo/test2/mod.rs
- src/big_repo/test2/harness/fixtures.rs
- src/big_repo/test2/convergence.rs
- src/big_repo/test2/edge.rs
- src/big_repo/runtime2/group_part_worker.rs
- src/big_repo/test2/restart.rs
- src/big_sync/test.rs
- src/big_repo/runtime2/doc_worker.rs
- src/big_repo/test2/notifications.rs
- src/big_sync/rpc.rs
- src/big_repo/keyhive.rs
- src/big_repo/runtime2/io.rs
- src/big_repo/test2/access_matrix.rs
- src/big_repo/runtime2/mod.rs
- src/big_repo/runtime2/messages.rs
- src/big_repo/runtime2/hub.rs
- src/big_sync/part_store.rs
- src/big_repo/test2/harness/topo.rs
- src/big_repo/sqlite_big_repo_store.rs
- src/big_repo/runtime.rs
| let permitted = guard | ||
| .doc_members | ||
| .get(&obj_id) | ||
| .and_then(|members| members.get(&principal)) | ||
| .map(|access| access.is_fetcher()) | ||
| .unwrap_or(false); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Object-notification permission default diverges from the SQLite store.
Here an object with no doc_members entry yields permitted = false, so no ObjectChanged is ever delivered. SqlitePartStore::publish (src/big_sync/part_store/sqlite.rs lines 544-552) defaults the same case to true and delivers. The two HostPartStore implementations should agree, otherwise contract tests and behavior differ depending on backend.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_sync/part_store/memory.rs` around lines 660 - 665, Update the
object-notification permission calculation in the `permitted` chain to default
to true when `doc_members` has no entry, matching `SqlitePartStore::publish`.
Preserve the existing `is_fetcher()` result when a member entry exists, so only
the missing-membership case changes.
| if live_part_ids.is_empty() { | ||
| tx.commit().await?; | ||
| if pending_part_ids.is_empty() { | ||
| self.publish(vec![SubEvent::ObjectChanged( | ||
| big_sync_core::rpc::ObjChangedWithoutPart { obj_id, payload }, | ||
| )]) | ||
| .await; | ||
| } else { | ||
| self.add_obj_to_parts(obj_id, pending_part_ids).await?; | ||
| } | ||
| return Ok(()); | ||
| } | ||
| assert!( | ||
| pending_part_ids.is_empty(), | ||
| "readable object cannot retain latent part memberships" | ||
| ); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Pending memberships are materialized outside the payload transaction.
pending_part_ids is read inside the transaction, but the transaction commits at line 932 and add_obj_to_parts (line 939) then opens a fresh one. A remove_obj_from_part interleaving between the two deletes the pending rows and then sees no live membership to remove, while this call re-materializes the very membership that was just revoked. The read and the materialization should share one transaction (or re-check pending rows inside add_obj_to_parts' transaction).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_sync/part_store/sqlite.rs` around lines 931 - 946, Keep pending
membership materialization atomic with the transaction that reads
pending_part_ids in the live_part_ids.is_empty() branch. Update add_obj_to_parts
or this branch so materialization uses the existing transaction, or re-reads and
validates pending rows within add_obj_to_parts’ transaction before inserting;
preserve the current publish behavior when no pending memberships remain.
| payload: added_payload | ||
| .clone() | ||
| .expect("visible membership requires added payload"), |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Any migration/backfill for added_payload_json?
rg -n 'added_payload_json' --type=rust -C2
# Any schema-version / migration machinery in the sqlite stores?
fd -a 'sqlite.rs|sqlite_big_repo_store.rs' src --exec rg -n 'ALTER TABLE|user_version|migrat' {}Repository: dman-os/townframe
Length of output: 155
🏁 Script executed:
#!/bin/bash
set -u
echo "== files =="
git ls-files 'src/**/*.rs' | rg 'sqlite|big_sync' | head -200
echo
echo "== added_payload_json occurrences =="
rg -n 'added_payload_json|added_payload' src --type=rust -C 3 -S || true
echo
echo "== sqlite.rs relevant function outlines =="
ast-grep outline src/big_sync/part_store/sqlite.rs --view compact 2>/dev/null | rg -n 'list_events|added_payload|init_schema|max|schema' || trueRepository: dman-os/townframe
Length of output: 13226
🏁 Script executed:
#!/bin/bash
set -u
echo "== sqlite.rs around list_events =="
sed -n '1500,1645p' src/big_sync/part_store/sqlite.rs | cat -n -v
echo
echo "== sqlite.rs schema around big_sync_members =="
sed -n '700,735p' src/big_sync/part_store/sqlite.rs | cat -n -v
echo
echo "== sqlite.rs schema for big_sync_objs =="
rg -n 'CREATE TABLE big_sync_objs|big_sync_objs|payload_json' src/big_sync/part_store/sqlite.rs -C 3 -S
echo
echo "== migration/version references in all relevant rust files =="
rg -n 'user_version|PRAGMA user_version|ALTER TABLE|sqlite_migration|migrations|Migrate|migrat|add_column' src --type=rust -C 2 -S || true
echo
echo "== added_payload_json references outside src =="
rg -n 'added_payload_json' -C 2 . --glob '!target/**' --glob '!node_modules/**' || trueRepository: dman-os/townframe
Length of output: 37860
Keep list_events graceful for nullable added_payload_json.
added_payload_json is nullable in big_sync_members, and this store has no migration/backfill for stale NULL rows. For Added events with added_payload_json IS NULL, fallback to the joined objs.payload_json here so loading existing SQLite-backed parts doesn’t panic on upgrade.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_sync/part_store/sqlite.rs` around lines 1605 - 1607, Update the
Added-event payload construction in list_events to handle nullable
added_payload_json without panicking: use added_payload when present, otherwise
fall back to the joined objs.payload_json value. Preserve the existing
expectation only where neither source is available, and ensure stale NULL rows
load gracefully.
| let mut nodes = Vec::with_capacity(node_count); | ||
| for peer_seed in 1..=(node_count as u8) { | ||
| let node = fixture.boot_node(Arc::clone(&world), peer_seed).await?; | ||
| nodes.push(Some(node)); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
node_count as u8 truncates and can silently boot an empty cluster.
With node_count == 256 the cast yields 0, so 1..=0 is empty, nodes is empty, and every downstream phase plus assert_cluster_alignment becomes vacuous — the stress test passes without running. A hard failure on an out-of-range count is the right behavior here.
🛡️ Proposed fix
- let mut nodes = Vec::with_capacity(node_count);
- for peer_seed in 1..=(node_count as u8) {
+ let node_count_u8 =
+ u8::try_from(node_count).expect("stress node_count must fit in a u8 peer seed");
+ assert!(node_count_u8 > 0, "stress node_count must be non-zero");
+ let mut nodes = Vec::with_capacity(node_count);
+ for peer_seed in 1..=node_count_u8 {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let mut nodes = Vec::with_capacity(node_count); | |
| for peer_seed in 1..=(node_count as u8) { | |
| let node = fixture.boot_node(Arc::clone(&world), peer_seed).await?; | |
| nodes.push(Some(node)); | |
| } | |
| let node_count_u8 = | |
| u8::try_from(node_count).expect("stress node_count must fit in a u8 peer seed"); | |
| assert!(node_count_u8 > 0, "stress node_count must be non-zero"); | |
| let mut nodes = Vec::with_capacity(node_count); | |
| for peer_seed in 1..=node_count_u8 { | |
| let node = fixture.boot_node(Arc::clone(&world), peer_seed).await?; | |
| nodes.push(Some(node)); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_sync/stress_support.rs` around lines 271 - 275, Validate node_count
before the boot loop so values exceeding the peer-seed range fail immediately
instead of being cast to u8. Update the loop around fixture.boot_node and
assert_cluster_alignment to use the validated range, preserving normal behavior
for supported counts and rejecting out-of-range counts explicitly.
| let final_settle_started_at = std::time::Instant::now(); | ||
| wait_for_cluster_settled( | ||
| &fixture, | ||
| &live_refs(&nodes), | ||
| Duration::from_secs(60), | ||
| "final:post-connect", | ||
| ) | ||
| .await?; | ||
| fixture | ||
| .wait_for_settled( | ||
| &live_refs(&nodes), | ||
| Duration::from_secs(60), | ||
| "final:post-connect", | ||
| ) | ||
| .await?; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Final settle ignores the caller's settle_timeout.
Every phase uses settle_timeout, but the final spanning-topology convergence — the longest-running wait — is hardcoded to 60s. Callers that pass a larger budget (e.g. the BigRepo stress fixture's SETTLE_TIMEOUT) will spuriously time out here.
🔧 Proposed fix
fixture
.wait_for_settled(
&live_refs(&nodes),
- Duration::from_secs(60),
+ settle_timeout,
"final:post-connect",
)
.await?;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let final_settle_started_at = std::time::Instant::now(); | |
| wait_for_cluster_settled( | |
| &fixture, | |
| &live_refs(&nodes), | |
| Duration::from_secs(60), | |
| "final:post-connect", | |
| ) | |
| .await?; | |
| fixture | |
| .wait_for_settled( | |
| &live_refs(&nodes), | |
| Duration::from_secs(60), | |
| "final:post-connect", | |
| ) | |
| .await?; | |
| let final_settle_started_at = std::time::Instant::now(); | |
| fixture | |
| .wait_for_settled( | |
| &live_refs(&nodes), | |
| settle_timeout, | |
| "final:post-connect", | |
| ) | |
| .await?; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_sync/stress_support.rs` around lines 750 - 757, Update the final
settle wait in the surrounding stress fixture flow to pass the caller-provided
settle_timeout instead of Duration::from_secs(60). Preserve the existing
final:post-connect label and wait_for_settled behavior while ensuring larger
caller budgets are honored.
| } | ||
| Ok(Err(error)) => { | ||
| runtime_evt_tx | ||
| .send(crate::runtime2::Runtime2Evt::FatalWorkerError { |
There was a problem hiding this comment.
todo: combine FatalWorkerError with DocWorkerStopped.
| runtime_evt_tx | ||
| .send(crate::runtime2::Runtime2Evt::FatalWorkerError { | ||
| doc_id: Some(doc_id), | ||
| context: "document worker failed", |
There was a problem hiding this comment.
todo: context field can just be part of the error string
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/big_repo/runtime2/doc_worker.rs (2)
869-877: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPartially decrypted blob sets can fail
load_incrementalwithMissingDepsand kill the worker.
try_decrypt_received_blobsdeliberately returns whatever it managed to decrypt when progress stalls (materialization_pending == true), butmaterialize_from_blobsmaps everyload_incrementalerror toferr!, which propagates out ofhandle_msgand becomes aFatalWorkerError.load_doc_snapshotalready special-casesAutomergeError::MissingDeps; the received-content path should treat it as "still pending" rather than fatal.🛡️ Sketch
for blob in blobs { - doc.load_incremental(blob) - .map_err(|error| ferr!("failed applying sync blob: {error}"))?; + match doc.load_incremental(blob) { + Ok(_) => {} + Err(automerge::AutomergeError::MissingDeps) => { + tracing::debug!(doc_id = %self.doc_id, "sync blob still missing dependencies"); + } + Err(error) => return Err(ferr!("failed applying sync blob: {error}")), + } }Also applies to: 897-902
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/doc_worker.rs` around lines 869 - 877, Update the received-content path in materialize_from_blobs to handle AutomergeError::MissingDeps from load_incremental as a pending materialization result rather than propagating it through ferr! as a fatal error. Preserve the partial blobs and pending state returned by try_decrypt_received_blobs, and mirror the existing MissingDeps handling in load_doc_snapshot; apply the same change to the additional load_incremental path noted in the comment.
662-674: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winThis
unreachable!is reachable and will take down the runtime.
DocIo::sedimentree_headsinsrc/big_repo/runtime2/native.rs(lines 384-453) has a path that returnsOk(Vec::new())after successfully hydrating non-empty durable heads: when the cache reported an empty frontier it logs the comparison warning and returns an empty vec regardless ofheads.len(). A live/materialized doc queried in that window hits exactly this branch and panics instead of degrading. Either fix the native fallback to return the derived heads, or downgrade this to thetracing::error!you already emit plus a returned error.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/doc_worker.rs` around lines 662 - 674, Fix the reachable invariant branch in the document worker’s state validation: avoid using unreachable! when materialized_heads is non-empty but sedimentree_heads is empty. Prefer correcting DocIo::sedimentree_heads to return the successfully hydrated derived heads instead of an empty vector; otherwise replace the panic with an error return while preserving the existing tracing::error! context.
🧹 Nitpick comments (1)
src/big_repo/runtime2/io.rs (1)
119-134: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDoc comment says "one serialized local document transition" but the method takes a batch.
Also worth documenting that the returned
FragmentRequestedset is the caller's obligation to service viastore_fragment, sincedoc_workerrelies on that contract.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/big_repo/runtime2/io.rs` around lines 119 - 134, Update the doc comment for persist_local_commits to describe persisting a batch of serialized local document transitions rather than one transition, and document that callers must service every returned FragmentRequested through store_fragment, as required by doc_worker.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/big_repo/runtime2/doc_worker.rs`:
- Around line 106-111: Update the Runtime2Evt::DocWorkerStopped send in the
worker-stop handling block to avoid relying on the racy is_closed() check and
expect. Handle a failed send as an ordinary shutdown outcome, matching the
existing Ok(Err(_)) arm below, while preserving the successful notification
path.
---
Outside diff comments:
In `@src/big_repo/runtime2/doc_worker.rs`:
- Around line 869-877: Update the received-content path in
materialize_from_blobs to handle AutomergeError::MissingDeps from
load_incremental as a pending materialization result rather than propagating it
through ferr! as a fatal error. Preserve the partial blobs and pending state
returned by try_decrypt_received_blobs, and mirror the existing MissingDeps
handling in load_doc_snapshot; apply the same change to the additional
load_incremental path noted in the comment.
- Around line 662-674: Fix the reachable invariant branch in the document
worker’s state validation: avoid using unreachable! when materialized_heads is
non-empty but sedimentree_heads is empty. Prefer correcting
DocIo::sedimentree_heads to return the successfully hydrated derived heads
instead of an empty vector; otherwise replace the panic with an error return
while preserving the existing tracing::error! context.
---
Nitpick comments:
In `@src/big_repo/runtime2/io.rs`:
- Around line 119-134: Update the doc comment for persist_local_commits to
describe persisting a batch of serialized local document transitions rather than
one transition, and document that callers must service every returned
FragmentRequested through store_fragment, as required by doc_worker.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: b58493da-539f-43c9-b67c-3fa0dc6475c9
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lockand included by**/*
📒 Files selected for processing (12)
Cargo.tomlsrc/big_repo/runtime.rssrc/big_repo/runtime2/doc_worker.rssrc/big_repo/runtime2/hub.rssrc/big_repo/runtime2/io.rssrc/big_repo/runtime2/lease.rssrc/big_repo/runtime2/messages.rssrc/big_repo/runtime2/mod.rssrc/big_repo/runtime2/native.rssrc/big_repo/test.rssrc/big_repo/test2/edge.rssrc/big_repo/test2/notifications.rs
🚧 Files skipped from review as they are similar to previous changes (6)
- Cargo.toml
- src/big_repo/runtime2/mod.rs
- src/big_repo/runtime2/hub.rs
- src/big_repo/runtime2/native.rs
- src/big_repo/test.rs
- src/big_repo/runtime.rs
| if matches!(&result, Ok(Ok(()))) && !runtime_evt_tx.is_closed() { | ||
| runtime_evt_tx | ||
| .send(Runtime2Evt::DocWorkerStopped { doc_id }) | ||
| .await | ||
| .expect("hub event channel checked open immediately before send"); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
is_closed() + await send is a TOCTOU; the expect can panic during shutdown.
The channel can close between the check and the completion of send().await (runtime stop races worker stop), turning an ordinary shutdown into a worker panic. Treat a closed hub channel here the same way the Ok(Err(_)) arm below already does.
🛡️ Proposed fix
- if matches!(&result, Ok(Ok(()))) && !runtime_evt_tx.is_closed() {
- runtime_evt_tx
- .send(Runtime2Evt::DocWorkerStopped { doc_id })
- .await
- .expect("hub event channel checked open immediately before send");
+ if matches!(&result, Ok(Ok(())))
+ && runtime_evt_tx
+ .send(Runtime2Evt::DocWorkerStopped { doc_id })
+ .await
+ .is_err()
+ {
+ tracing::debug!(%doc_id, "runtime stopped before doc worker stop event");
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if matches!(&result, Ok(Ok(()))) && !runtime_evt_tx.is_closed() { | |
| runtime_evt_tx | |
| .send(Runtime2Evt::DocWorkerStopped { doc_id }) | |
| .await | |
| .expect("hub event channel checked open immediately before send"); | |
| } | |
| if matches!(&result, Ok(Ok(()))) | |
| && runtime_evt_tx | |
| .send(Runtime2Evt::DocWorkerStopped { doc_id }) | |
| .await | |
| .is_err() | |
| { | |
| tracing::debug!(%doc_id, "runtime stopped before doc worker stop event"); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/big_repo/runtime2/doc_worker.rs` around lines 106 - 111, Update the
Runtime2Evt::DocWorkerStopped send in the worker-stop handling block to avoid
relying on the racy is_closed() check and expect. Handle a failed send as an
ordinary shutdown outcome, matching the existing Ok(Err(_)) arm below, while
preserving the successful notification path.
| init_schema(&sql.write_pool, bucket_depth).await?; | ||
| let scope_key = scope_key.into(); | ||
| let scope_id = Self::ensure_scope_id(&sql.write_pool, &scope_key).await?; | ||
| // Rehydrate doc_members_cache from persisted syncable rows. |
There was a problem hiding this comment.
issue: do we need an in memory cache? that might be prohibitive on large parts? why not query sqlite directly?
| @@ -21,13 +21,117 @@ use sqlx_utils_rs::SqlCtx; | |||
| #[cfg(test)] | |||
| use uuid::Uuid; | |||
|
|
|||
| const SUB_REPLAYING_CLEAN: u8 = 0; | |||
There was a problem hiding this comment.
issue: use enums here with conversion methods
| sub.try_send(evt.clone()).is_ok() | ||
| }); | ||
| } | ||
| if !matches!(object_event, SubEvent::ReplayComplete) { |
There was a problem hiding this comment.
todo: ReplayComplete will never show up here, might as well be an assert
There was a problem hiding this comment.
can we dedup shared stuff with the base sqlite impl?
| hidden_parts: Arc::new(config.hidden_parts), | ||
| doc_members_cache: Arc::new(std::sync::RwLock::new(doc_members)), | ||
| }; | ||
| store.init_subduction_schema().await?; |
There was a problem hiding this comment.
issue: this should just be part of init_schema
| } | ||
| Runtime2Cmd::CheckSedimentreeResident { doc_id, resp } => { | ||
| let sedimentree_id = sedimentree_core::id::SedimentreeId::new(doc_id.into_bytes()); | ||
| self.spawn_background(F::check_sedimentree( |
There was a problem hiding this comment.
issue: check_sedimentree should be renamed to contains_sedimentree
| || repo.runtime.contains_sedimentree_id(doc_id).await?; | ||
| let has_local_doc_state = match futures::future::select( | ||
| core::pin::pin!(repo.runtime.has_doc_worker(doc_id)), | ||
| core::pin::pin!(repo.runtime.contains_sedimentree_id(doc_id)), |
There was a problem hiding this comment.
issue: this should be renamed to contains_sedimentree. what's more, maybe the runtime tiself should combine the has_doc_worker and contains_sedimentree checks instead of a select
| round_id, | ||
| result, | ||
| } => { | ||
| self.finish_keyhive_sync_after_cache(peer_id, round_id, result)?; |
There was a problem hiding this comment.
question: do we really need to wait for cache refresh after syncing from remote? I'm guessing this is for testing so that after a sync_keyhive call, we can rely on the subduction_keyhive cache to be fresh? but local queries on the keyhive won't be out of date, right? only subduction policy which relies on the cache? maybe we move this into quiscence, maybe we should rework our tests so that sync_keyhive is followed by quiscence before we do sync_doc so we can remove this from the real sync_keyhive path?
There was a problem hiding this comment.
issue: do we use still? if not, let's yeet
There was a problem hiding this comment.
todo: retire the old runtime code, we need to move some of the runtime2 used types to it's own module tho
1 participant
Based on top of #30
Summary by CodeRabbit