refactor(sessions): Part 1 - store actor & plumbing - #744
Conversation
📝 WalkthroughWalkthroughThe change adds an asynchronous session store with predicate-based record handles, updates the session manager to use it for lifecycle operations, and hardens ChangesSession store migration
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Manager
participant StoreHandle
participant SessionRecordHandle
participant DiskLoader
Manager->>StoreHandle: find RecordPredicate
StoreHandle->>DiskLoader: resolve session key
StoreHandle-->>Manager: SessionRecordHandle
Manager->>SessionRecordHandle: record, write, or delete
SessionRecordHandle->>DiskLoader: perform record operation
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
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 (1)
crates/minimald/src/sessions.rs (1)
318-339: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winA single undeletable orphan Pending record fails the entire daemon startup.
handle.delete().await?propagates any failure straight out ofreap_orphan_pendingand then out ofManager::init— so one Pending session whose on-disk delete fails (permission issue, read-only mount, etc.) blocks the daemon from starting at all, even though every other session is fine. TheSubmitVerdictresume-failure branch a few hundred lines down (657-668) already establishes the "best-effort delete + log a warning, don't fail the caller" pattern for the same class of problem — worth applying the same resilience here.🛡️ Suggested fix: don't let one failed delete abort the whole reap pass
async fn reap_orphan_pending(store: &StoreHandle) -> Result<(), std::io::Error> { let mut reaped = 0u64; for handle in store.handles().await? { if handle.record().await?.status == sessions::SessionStatus::Pending { let id = *handle.id(); if let Err(e) = handle.delete().await { tracing::warn!( session_id = %id, error = %e, "failed to reap orphan Pending session on startup; will retry next startup", ); continue; } tracing::info!( session_id = %id, "reaped orphan Pending session on startup", ); reaped += 1; } } if reaped > 0 { tracing::info!(count = reaped, "orphan Pending reap complete"); } 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 `@crates/minimald/src/sessions.rs` around lines 318 - 339, Update Manager::reap_orphan_pending to handle each handle.delete() failure locally: log a warning with the session ID and error, then continue reaping remaining sessions instead of propagating the deletion error. Only increment the reaped count and emit the success log after a successful deletion; preserve propagation of errors from listing handles or reading records.
🧹 Nitpick comments (1)
crates/minimald/src/store.rs (1)
80-137: 🚀 Performance & Scalability | 🔵 TrivialBlocking filesystem I/O runs directly on the async actor task.
Every branch in
handle_message(self.store.get/save/delete/create/keys) does synchronousstd::fsI/O — includingfile.sync_all()inDiskLoader::write_record/flush_index— inline inside thisasync fn, which runs on the Tokio runtime rather than a dedicated blocking pool. Tokio's own guidance is that tasks should generally not perform system calls or other operations that could block a thread, as this would prevent other tasks running on the same thread from executing as well, and recommends offloading such work viaspawn_blocking. Since every session read/write/delete now funnels through this single actor's serialized mainloop, a slow fsync or disk hiccup stalls that worker thread for the duration, delaying all other in-flightStorerequests (and anything else scheduled on the same worker in a multi-thread runtime).This mirrors the pre-existing behavior of the old synchronous
Manager, so it's not a regression, but now that record I/O is centralized inStore, it's worth consideringtokio::task::spawn_blocking(movingDiskLoaderownership into the blocking closure and back) if session counts/record I/O become a bottleneck.🤖 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 `@crates/minimald/src/store.rs` around lines 80 - 137, Offload the synchronous filesystem operations performed by Store to Tokio’s blocking pool instead of executing them inline in the async actor. Update mainloop/handle_message and the self.store operations used by Handles, Create, Find, SessionGet, SessionObject, SessionWrite, and SessionDelete to run through spawn_blocking while preserving serialized access, response types, and DiskLoader ownership across the blocking work.
🤖 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 `@crates/sessions/src/store.rs`:
- Around line 47-50: Update the NotFound documentation for Self::save to
describe both stale-key cases: the short missing from the index and the short
being reallocated to a different session. Preserve the existing explanation that
this prevents stale writes from resurrecting or overwriting sessions, and ensure
the Self::load cross-reference accurately matches save’s documented semantics.
---
Outside diff comments:
In `@crates/minimald/src/sessions.rs`:
- Around line 318-339: Update Manager::reap_orphan_pending to handle each
handle.delete() failure locally: log a warning with the session ID and error,
then continue reaping remaining sessions instead of propagating the deletion
error. Only increment the reaped count and emit the success log after a
successful deletion; preserve propagation of errors from listing handles or
reading records.
---
Nitpick comments:
In `@crates/minimald/src/store.rs`:
- Around line 80-137: Offload the synchronous filesystem operations performed by
Store to Tokio’s blocking pool instead of executing them inline in the async
actor. Update mainloop/handle_message and the self.store operations used by
Handles, Create, Find, SessionGet, SessionObject, SessionWrite, and
SessionDelete to run through spawn_blocking while preserving serialized access,
response types, and DiskLoader ownership across the blocking work.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3d434a11-65da-4c5e-bb2f-6543c8bdaf4b
📒 Files selected for processing (5)
crates/minimald/src/lib.rscrates/minimald/src/sessions.rscrates/minimald/src/sessions/composables.rscrates/minimald/src/store.rscrates/sessions/src/store.rs
sessions::storehandlesget()of a deleted key gracefullyminimald::Storeactor to mediate read/write of session recordsSessionRecordHandleto drive the store actor to read/write/delete a specific session recordsessionsactor to initialize and use newstoreactor.Part 2 will refactor the
sessionactor to own everything about a specific session, including implementing a state machine for when the session is being built. Hopefullly this will makesessionsa dumb router again.Summary by CodeRabbit
New Features
Bug Fixes