Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: make WorkerId the standalone primary identity

Browse Source
This commit is contained in:
Hare
2026-08-31 14:37:46 +09:00
parent bde1dea2a5
commit e7f4c6864f
16 changed files with 444 additions and 437 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/standalone/src/host.rs
+71 -69
View File
@@ -5,7 +5,7 @@ use agen::llm_client::client::LlmClient;
use client::Client;
use client::transport::in_process::{Peer as InProcessPeer, Socket as InProcessSocket};
use protocol::stream::{decode_method, encode_event};
use protocol::{Event, Method};
use protocol::{Event, Method, WorkerId};
use session_store::{
CombinedStore, FsStore, FsWorkerStore, WorkerActiveSegmentRef, WorkerMetadataStore,
};
@@ -20,14 +20,14 @@ use worker::{BootstrappedWorker, WorkerError, WorkerFilesystemAuthority, WorkerW
use crate::launch::ResolvedStandaloneLaunch;
use crate::store::{
StaleLeasePolicy, StandaloneSessionId, StandaloneSessionLease, StandaloneSessionRecord,
StandaloneSessionStore, StandaloneShutdownReason, StandaloneStoreError,
StaleLeasePolicy, StandaloneShutdownReason, StandaloneStoreError, StandaloneWorkerLease,
StandaloneWorkerRecord, StandaloneWorkerStore,
};
const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
type StandaloneBackingStore = CombinedStore<FsStore, FsWorkerStore>;
/// One client-owned top-level Worker and its standalone session authority.
/// One client-owned top-level Worker and its standalone Worker authority.
///
/// The host deliberately exposes the existing typed Worker protocol rather than owning an
/// HTTP/WebSocket server or creating Runtime/Workspace/Ticket/Workdir domain records.
@@ -35,21 +35,21 @@ pub struct StandaloneHost {
handle: worker::WorkerHandle,
shutdown: Option<worker::controller::ShutdownReceiver>,
shutdown_timeout: Duration,
store: StandaloneSessionStore,
store: StandaloneWorkerStore,
worker_store: FsWorkerStore,
record: StandaloneSessionRecord,
lease: Option<StandaloneSessionLease>,
record: StandaloneWorkerRecord,
lease: Option<StandaloneWorkerLease>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
pub enum StandaloneStartupError {
#[error("the standalone state store could not be opened or validated")]
StateStore,
#[error("the standalone session is already active")]
SessionActive,
#[error("the standalone session lease cannot be observed safely; recovery is rejected")]
#[error("the standalone Worker is already active")]
WorkerActive,
#[error("the standalone Worker lease cannot be observed safely; recovery is rejected")]
LeaseLivenessUnknown,
#[error("the standalone session working directory is unavailable or changed")]
#[error("the standalone Worker working directory is unavailable or changed")]
WorkingDirectoryUnavailable,
#[error("the resolved Worker configuration or persisted history is invalid")]
WorkerConfiguration,
@@ -67,7 +67,7 @@ pub enum StandaloneShutdownError {
DeadlineExceeded,
#[error("the standalone Worker shutdown confirmation was lost")]
ConfirmationLost,
#[error("the standalone session final state could not be committed")]
#[error("the standalone Worker final state could not be committed")]
StateStore,
}
@@ -87,22 +87,24 @@ impl StandaloneHost {
}
async fn start_with_optional_model_client(
mut launch: ResolvedStandaloneLaunch,
launch: ResolvedStandaloneLaunch,
model_client: Option<Box<dyn LlmClient>>,
) -> Result<Self, StandaloneStartupError> {
let store = StandaloneSessionStore::open(&launch.state_dir)
.map_err(classify_store_startup_error)?;
let store =
StandaloneWorkerStore::open(&launch.state_dir).map_err(classify_store_startup_error)?;
let allocation = store
.allocate(&launch.cwd, StaleLeasePolicy::Reject)
.map_err(classify_store_startup_error)?;
let id = allocation.id();
let worker_id = allocation.worker_id();
// The standalone session ID is the local identity. A unique internal Worker name avoids
// process-global allocation collisions without creating a Runtime/Workspace Worker ID.
launch.profile.manifest.worker.name = format!("standalone-{id}");
// WorkerId is the stable identity. The current Worker store remains
// name-keyed, so keep its derived storage key separate from the
// user-facing profile name.
let manifest = launch.profile.manifest.clone();
let worker_name = manifest.worker.name.clone();
let (backing_store, worker_store) = match backing_store(&store, id) {
let storage_key = format!("standalone-{worker_id}");
let mut bootstrap_manifest = manifest.clone();
bootstrap_manifest.worker.name = storage_key.clone();
let (backing_store, worker_store) = match backing_store(&store, worker_id) {
Ok(stores) => stores,
Err(error) => {
let _ = store.abandon_allocation(allocation);
@@ -112,10 +114,10 @@ impl StandaloneHost {
let filesystem_authority =
WorkerFilesystemAuthority::local(launch.cwd.clone(), launch.cwd.clone());
let workspace_context = WorkerWorkspaceContext::local_filesystem(None);
let runtime_base = store.runtime_dir(id);
let runtime_base = store.runtime_dir(worker_id);
let mut bootstrap = WorkerBootstrap::new(
manifest.clone(),
bootstrap_manifest,
backing_store,
launch.prompt_catalog,
workspace_context,
@@ -133,7 +135,7 @@ impl StandaloneHost {
return Err(classify_startup_error(error));
}
};
let active = match active_pointer(&worker_store, &worker_name) {
let active = match active_pointer(&worker_store, &storage_key) {
Ok(active) => active,
Err(error) => {
stop_started_worker(started).await;
@@ -141,16 +143,20 @@ impl StandaloneHost {
return Err(error);
}
};
let record =
match store.commit_created(&allocation, manifest, active.session_id, active.segment_id)
{
Ok(record) => record,
Err(_) => {
stop_started_worker(started).await;
let _ = store.abandon_allocation(allocation);
return Err(StandaloneStartupError::StateStore);
}
};
let record = match store.commit_created(
&allocation,
manifest,
storage_key,
active.session_id,
active.segment_id,
) {
Ok(record) => record,
Err(_) => {
stop_started_worker(started).await;
let _ = store.abandon_allocation(allocation);
return Err(StandaloneStartupError::StateStore);
}
};
Ok(Self::from_started(
started,
store,
@@ -162,50 +168,46 @@ impl StandaloneHost {
pub async fn restore(
state_dir: PathBuf,
session_id: StandaloneSessionId,
worker_id: WorkerId,
) -> Result<Self, StandaloneStartupError> {
Self::restore_with_optional_model_client(state_dir, session_id, None).await
Self::restore_with_optional_model_client(state_dir, worker_id, None).await
}
pub async fn restore_with_model_client<C>(
state_dir: PathBuf,
session_id: StandaloneSessionId,
worker_id: WorkerId,
model_client: C,
) -> Result<Self, StandaloneStartupError>
where
C: LlmClient + 'static,
{
Self::restore_with_optional_model_client(
state_dir,
session_id,
Some(Box::new(model_client)),
)
.await
Self::restore_with_optional_model_client(state_dir, worker_id, Some(Box::new(model_client)))
.await
}
async fn restore_with_optional_model_client(
state_dir: PathBuf,
session_id: StandaloneSessionId,
worker_id: WorkerId,
model_client: Option<Box<dyn LlmClient>>,
) -> Result<Self, StandaloneStartupError> {
let store =
StandaloneSessionStore::open(state_dir).map_err(classify_store_startup_error)?;
let store = StandaloneWorkerStore::open(state_dir).map_err(classify_store_startup_error)?;
let record = store
.load(session_id)
.load(worker_id)
.map_err(classify_store_startup_error)?;
record.cwd.verify().map_err(classify_store_startup_error)?;
let lease = store
.acquire_lease(session_id, StaleLeasePolicy::Recover)
.acquire_lease(worker_id, StaleLeasePolicy::Recover)
.map_err(classify_store_startup_error)?;
let (backing_store, worker_store) = backing_store(&store, session_id)?;
let worker_name = record.worker_name.clone();
let manifest = record.manifest.clone();
let (backing_store, worker_store) = backing_store(&store, worker_id)?;
let storage_key = record.storage_key.clone();
let mut manifest = record.manifest.clone();
manifest.worker.name = storage_key.clone();
let filesystem_authority = WorkerFilesystemAuthority::local(
record.cwd.canonical_path.clone(),
record.cwd.canonical_path.clone(),
);
let workspace_context = WorkerWorkspaceContext::local_filesystem(None);
let runtime_base = store.runtime_dir(session_id);
let runtime_base = store.runtime_dir(worker_id);
let mut bootstrap = WorkerBootstrap::new(
manifest,
@@ -220,11 +222,11 @@ impl StandaloneHost {
bootstrap = bootstrap.with_model_client(model_client);
}
let prepared = bootstrap
.prepare_restored(&worker_name)
.prepare_restored(&storage_key)
.await
.map_err(classify_startup_error)?;
let started = prepared.start().await.map_err(classify_startup_error)?;
let active = match active_pointer(&worker_store, &worker_name) {
let active = match active_pointer(&worker_store, &storage_key) {
Ok(active) => active,
Err(error) => {
stop_started_worker(started).await;
@@ -251,10 +253,10 @@ impl StandaloneHost {
fn from_started(
started: BootstrappedWorker,
store: StandaloneSessionStore,
store: StandaloneWorkerStore,
worker_store: FsWorkerStore,
record: StandaloneSessionRecord,
lease: StandaloneSessionLease,
record: StandaloneWorkerRecord,
lease: StandaloneWorkerLease,
) -> Self {
Self {
handle: started.handle,
@@ -268,12 +270,12 @@ impl StandaloneHost {
}
#[must_use]
pub fn session_id(&self) -> StandaloneSessionId {
self.record.session_id
pub fn worker_id(&self) -> WorkerId {
self.record.worker_id
}
#[must_use]
pub fn record(&self) -> &StandaloneSessionRecord {
pub fn record(&self) -> &StandaloneWorkerRecord {
&self.record
}
@@ -310,7 +312,7 @@ impl StandaloneHost {
return Err(StandaloneShutdownError::DeadlineExceeded);
}
}
let active = match active_pointer(&self.worker_store, &self.record.worker_name) {
let active = match active_pointer(&self.worker_store, &self.record.storage_key) {
Ok(active) => active,
Err(_) => {
self.retain_lease();
@@ -451,12 +453,12 @@ async fn send_protocol_event(peer: &InProcessPeer, event: Event) -> bool {
}
fn backing_store(
store: &StandaloneSessionStore,
id: StandaloneSessionId,
store: &StandaloneWorkerStore,
worker_id: WorkerId,
) -> Result<(StandaloneBackingStore, FsWorkerStore), StandaloneStartupError> {
let session_store =
FsStore::new(store.session_log_dir(id)).map_err(|_| StandaloneStartupError::StateStore)?;
let worker_store = FsWorkerStore::new(store.worker_metadata_dir(id))
let session_store = FsStore::new(store.sessions_dir(worker_id))
.map_err(|_| StandaloneStartupError::StateStore)?;
let worker_store = FsWorkerStore::new(store.worker_metadata_dir(worker_id))
.map_err(|_| StandaloneStartupError::StateStore)?;
Ok((
CombinedStore::new(session_store, worker_store.clone()),
@@ -466,10 +468,10 @@ fn backing_store(
fn active_pointer(
worker_store: &FsWorkerStore,
worker_name: &str,
storage_key: &str,
) -> Result<WorkerActiveSegmentRef, StandaloneStartupError> {
worker_store
.read_by_name(worker_name)
.read_by_name(storage_key)
.map_err(|_| StandaloneStartupError::StateStore)?
.and_then(|metadata| metadata.active)
.ok_or(StandaloneStartupError::StateStore)
@@ -482,7 +484,7 @@ async fn stop_started_worker(started: BootstrappedWorker) {
fn classify_store_startup_error(error: StandaloneStoreError) -> StandaloneStartupError {
match error {
StandaloneStoreError::SessionLeased(_) => StandaloneStartupError::SessionActive,
StandaloneStoreError::WorkerLeased(_) => StandaloneStartupError::WorkerActive,
StandaloneStoreError::LeaseLivenessUnknown(_) => {
StandaloneStartupError::LeaseLivenessUnknown
}
crates/standalone/src/lib.rs
+3 -3
View File
@@ -10,8 +10,8 @@ pub mod store;
pub use host::{StandaloneHost, StandaloneShutdownError, StandaloneStartupError};
pub use launch::{ResolvedStandaloneLaunch, StandaloneLaunchConfig, StandaloneLaunchError};
pub use protocol::WorkerId;
pub use store::{
StaleLeasePolicy, StandaloneCwdIdentity, StandaloneListScope, StandaloneSessionId,
StandaloneSessionRecord, StandaloneSessionStatus, StandaloneSessionStore,
StandaloneShutdownReason, StandaloneStoreError,
StaleLeasePolicy, StandaloneCwdIdentity, StandaloneListScope, StandaloneShutdownReason,
StandaloneStoreError, StandaloneWorkerRecord, StandaloneWorkerStatus, StandaloneWorkerStore,
};
crates/standalone/src/store.rs
+106 -143
View File
@@ -1,12 +1,11 @@
use std::fmt;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::time::{SystemTime, UNIX_EPOCH};
use fs4::fs_std::FileExt;
use manifest::WorkerManifest;
use protocol::WorkerId;
use serde::{Deserialize, Serialize};
use session_store::{SegmentId, SessionId};
use thiserror::Error;
@@ -16,47 +15,10 @@ const RECORD_FILE: &str = "record.json";
const COMMIT_MARKER: &str = "commit.pending";
const LEASE_FILE: &str = "lease.json";
const LEASE_LOCK_FILE: &str = "lease.lock";
const SESSION_DIR: &str = "session";
const SESSIONS_DIR: &str = "sessions";
const WORKER_DIR: &str = "worker";
const SCHEMA_VERSION: u32 = 1;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct StandaloneSessionId(Uuid);
impl StandaloneSessionId {
#[must_use]
pub fn new() -> Self {
Self(Uuid::now_v7())
}
#[must_use]
pub fn short(self) -> String {
let simple = self.0.simple().to_string();
simple[simple.len() - 12..].to_string()
}
}
impl Default for StandaloneSessionId {
fn default() -> Self {
Self::new()
}
}
impl fmt::Display for StandaloneSessionId {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(formatter)
}
}
impl FromStr for StandaloneSessionId {
type Err = uuid::Error;
fn from_str(value: &str) -> Result<Self, Self::Err> {
Uuid::parse_str(value).map(Self)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StandaloneCwdIdentity {
pub canonical_path: PathBuf,
@@ -100,7 +62,7 @@ impl StandaloneCwdIdentity {
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StandaloneSessionStatus {
pub enum StandaloneWorkerStatus {
Active,
Stopped,
}
@@ -115,17 +77,20 @@ pub enum StandaloneShutdownReason {
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StandaloneSessionRecord {
pub struct StandaloneWorkerRecord {
pub schema_version: u32,
pub revision: u64,
pub session_id: StandaloneSessionId,
pub worker_id: WorkerId,
/// User-facing Worker name resolved from the profile.
pub worker_name: String,
/// Internal key used by the current name-keyed Worker store.
pub storage_key: String,
pub cwd: StandaloneCwdIdentity,
pub manifest: WorkerManifest,
pub active_session_id: SessionId,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub active_segment_id: Option<SegmentId>,
pub status: StandaloneSessionStatus,
pub status: StandaloneWorkerStatus,
pub created_at_unix_ms: u64,
pub updated_at_unix_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -145,11 +110,11 @@ pub enum StaleLeasePolicy {
}
#[derive(Debug, Clone)]
pub struct StandaloneSessionStore {
pub struct StandaloneWorkerStore {
root: PathBuf,
}
impl StandaloneSessionStore {
impl StandaloneWorkerStore {
pub fn open(root: impl Into<PathBuf>) -> Result<Self, StandaloneStoreError> {
let root = root.into();
fs::create_dir_all(&root).map_err(StandaloneStoreError::Io)?;
@@ -171,35 +136,41 @@ impl StandaloneSessionStore {
&self,
cwd: impl AsRef<Path>,
policy: StaleLeasePolicy,
) -> Result<StandaloneSessionAllocation, StandaloneStoreError> {
let id = StandaloneSessionId::new();
) -> Result<StandaloneWorkerAllocation, StandaloneStoreError> {
let worker_id = WorkerId::now_v7();
let cwd = StandaloneCwdIdentity::capture(cwd)?;
let dir = self.session_dir(id);
let dir = self.worker_dir(worker_id);
fs::create_dir(&dir).map_err(StandaloneStoreError::Io)?;
fs::create_dir(dir.join(SESSION_DIR)).map_err(StandaloneStoreError::Io)?;
fs::create_dir(dir.join(SESSIONS_DIR)).map_err(StandaloneStoreError::Io)?;
fs::create_dir(dir.join(WORKER_DIR)).map_err(StandaloneStoreError::Io)?;
let lease = self.acquire_lease(id, policy)?;
Ok(StandaloneSessionAllocation { id, cwd, lease })
let lease = self.acquire_lease(worker_id, policy)?;
Ok(StandaloneWorkerAllocation {
worker_id,
cwd,
lease,
})
}
pub fn commit_created(
&self,
allocation: &StandaloneSessionAllocation,
allocation: &StandaloneWorkerAllocation,
manifest: WorkerManifest,
storage_key: String,
active_session_id: SessionId,
active_segment_id: Option<SegmentId>,
) -> Result<StandaloneSessionRecord, StandaloneStoreError> {
) -> Result<StandaloneWorkerRecord, StandaloneStoreError> {
let now = now_unix_ms()?;
let record = StandaloneSessionRecord {
let record = StandaloneWorkerRecord {
schema_version: SCHEMA_VERSION,
revision: 1,
session_id: allocation.id,
worker_id: allocation.worker_id,
worker_name: manifest.worker.name.clone(),
storage_key,
cwd: allocation.cwd.clone(),
manifest,
active_session_id,
active_segment_id,
status: StandaloneSessionStatus::Active,
status: StandaloneWorkerStatus::Active,
created_at_unix_ms: now,
updated_at_unix_ms: now,
shutdown_reason: None,
@@ -208,22 +179,19 @@ impl StandaloneSessionStore {
Ok(record)
}
pub fn load(
&self,
id: StandaloneSessionId,
) -> Result<StandaloneSessionRecord, StandaloneStoreError> {
let dir = self.session_dir(id);
pub fn load(&self, id: WorkerId) -> Result<StandaloneWorkerRecord, StandaloneStoreError> {
let dir = self.worker_dir(id);
if dir.join(COMMIT_MARKER).exists() {
return Err(StandaloneStoreError::IncompleteCommit(id));
}
let bytes = fs::read(dir.join(RECORD_FILE)).map_err(|error| {
if error.kind() == io::ErrorKind::NotFound {
StandaloneStoreError::SessionNotFound(id)
StandaloneStoreError::WorkerNotFound(id)
} else {
StandaloneStoreError::Io(error)
}
})?;
let record: StandaloneSessionRecord = serde_json::from_slice(&bytes)
let record: StandaloneWorkerRecord = serde_json::from_slice(&bytes)
.map_err(|source| StandaloneStoreError::CorruptRecord { id, source })?;
if record.schema_version > SCHEMA_VERSION {
return Err(StandaloneStoreError::NewerSchema {
@@ -232,7 +200,7 @@ impl StandaloneSessionStore {
supported: SCHEMA_VERSION,
});
}
if record.schema_version != SCHEMA_VERSION || record.session_id != id {
if record.schema_version != SCHEMA_VERSION || record.worker_id != id {
return Err(StandaloneStoreError::InvalidRecord(id));
}
Ok(record)
@@ -243,7 +211,7 @@ impl StandaloneSessionStore {
cwd: impl AsRef<Path>,
scope: StandaloneListScope,
limit: usize,
) -> Result<Vec<StandaloneSessionRecord>, StandaloneStoreError> {
) -> Result<Vec<StandaloneWorkerRecord>, StandaloneStoreError> {
let current_cwd = (scope == StandaloneListScope::CurrentCwd)
.then(|| StandaloneCwdIdentity::capture(cwd))
.transpose()?;
@@ -269,12 +237,7 @@ impl StandaloneSessionStore {
right
.updated_at_unix_ms
.cmp(&left.updated_at_unix_ms)
.then_with(|| {
right
.session_id
.to_string()
.cmp(&left.session_id.to_string())
})
.then_with(|| right.worker_id.to_string().cmp(&left.worker_id.to_string()))
});
records.truncate(limit);
Ok(records)
@@ -282,10 +245,10 @@ impl StandaloneSessionStore {
pub fn acquire_lease(
&self,
id: StandaloneSessionId,
id: WorkerId,
policy: StaleLeasePolicy,
) -> Result<StandaloneSessionLease, StandaloneStoreError> {
let dir = self.session_dir(id);
) -> Result<StandaloneWorkerLease, StandaloneStoreError> {
let dir = self.worker_dir(id);
let path = dir.join(LEASE_FILE);
let _guard = LeaseMutationGuard::acquire(&dir)?;
let lease = LeaseRecord::current()?;
@@ -296,7 +259,7 @@ impl StandaloneSessionStore {
file.write_all(b"\n").map_err(StandaloneStoreError::Io)?;
file.sync_all().map_err(StandaloneStoreError::Io)?;
sync_directory(&dir)?;
return Ok(StandaloneSessionLease {
return Ok(StandaloneWorkerLease {
path,
lease_id: lease.lease_id,
released: false,
@@ -306,7 +269,7 @@ impl StandaloneSessionStore {
let existing = read_lease(&path, id)?;
match existing.liveness() {
LeaseLiveness::Live => {
return Err(StandaloneStoreError::SessionLeased(id));
return Err(StandaloneStoreError::WorkerLeased(id));
}
LeaseLiveness::Unknown => {
return Err(StandaloneStoreError::LeaseLivenessUnknown(id));
@@ -326,16 +289,16 @@ impl StandaloneSessionStore {
pub fn update_active_pointer(
&self,
record: &StandaloneSessionRecord,
record: &StandaloneWorkerRecord,
active_session_id: SessionId,
active_segment_id: Option<SegmentId>,
) -> Result<StandaloneSessionRecord, StandaloneStoreError> {
) -> Result<StandaloneWorkerRecord, StandaloneStoreError> {
let mut next = record.clone();
next.revision = next.revision.saturating_add(1);
next.updated_at_unix_ms = now_unix_ms()?;
next.active_session_id = active_session_id;
next.active_segment_id = active_segment_id;
next.status = StandaloneSessionStatus::Active;
next.status = StandaloneWorkerStatus::Active;
next.shutdown_reason = None;
self.commit_record(Some(record.revision), &next)?;
Ok(next)
@@ -343,73 +306,73 @@ impl StandaloneSessionStore {
pub fn mark_stopped(
&self,
record: &StandaloneSessionRecord,
record: &StandaloneWorkerRecord,
active_session_id: SessionId,
active_segment_id: Option<SegmentId>,
reason: StandaloneShutdownReason,
) -> Result<StandaloneSessionRecord, StandaloneStoreError> {
) -> Result<StandaloneWorkerRecord, StandaloneStoreError> {
let mut next = record.clone();
next.revision = next.revision.saturating_add(1);
next.updated_at_unix_ms = now_unix_ms()?;
next.active_session_id = active_session_id;
next.active_segment_id = active_segment_id;
next.status = StandaloneSessionStatus::Stopped;
next.status = StandaloneWorkerStatus::Stopped;
next.shutdown_reason = Some(reason);
self.commit_record(Some(record.revision), &next)?;
Ok(next)
}
pub fn delete(&self, id: StandaloneSessionId) -> Result<(), StandaloneStoreError> {
pub fn delete(&self, id: WorkerId) -> Result<(), StandaloneStoreError> {
let record = self.load(id)?;
if record.status != StandaloneSessionStatus::Stopped {
if record.status != StandaloneWorkerStatus::Stopped {
return Err(StandaloneStoreError::DeleteActive(id));
}
let session_dir = self.session_dir(id);
let _guard = LeaseMutationGuard::acquire(&session_dir)?;
let lease_path = session_dir.join(LEASE_FILE);
let worker_dir = self.worker_dir(id);
let _guard = LeaseMutationGuard::acquire(&worker_dir)?;
let lease_path = worker_dir.join(LEASE_FILE);
if lease_path.exists() {
let lease = read_lease(&lease_path, id)?;
return Err(match lease.liveness() {
LeaseLiveness::Live => StandaloneStoreError::SessionLeased(id),
LeaseLiveness::Live => StandaloneStoreError::WorkerLeased(id),
LeaseLiveness::Stale => StandaloneStoreError::StaleLease(id),
LeaseLiveness::Unknown => StandaloneStoreError::LeaseLivenessUnknown(id),
});
}
fs::remove_dir_all(self.session_dir(id)).map_err(StandaloneStoreError::Io)?;
fs::remove_dir_all(self.worker_dir(id)).map_err(StandaloneStoreError::Io)?;
sync_directory(&self.root)
}
#[must_use]
pub fn session_log_dir(&self, id: StandaloneSessionId) -> PathBuf {
self.session_dir(id).join(SESSION_DIR)
pub fn sessions_dir(&self, id: WorkerId) -> PathBuf {
self.worker_dir(id).join(SESSIONS_DIR)
}
#[must_use]
pub fn worker_metadata_dir(&self, id: StandaloneSessionId) -> PathBuf {
self.session_dir(id).join(WORKER_DIR)
pub fn worker_metadata_dir(&self, id: WorkerId) -> PathBuf {
self.worker_dir(id).join(WORKER_DIR)
}
#[must_use]
pub(crate) fn runtime_dir(&self, id: StandaloneSessionId) -> PathBuf {
self.session_dir(id).join("runtime")
pub(crate) fn runtime_dir(&self, id: WorkerId) -> PathBuf {
self.worker_dir(id).join("runtime")
}
pub(crate) fn abandon_allocation(
&self,
allocation: StandaloneSessionAllocation,
allocation: StandaloneWorkerAllocation,
) -> Result<(), StandaloneStoreError> {
let id = allocation.id;
let worker_id = allocation.worker_id;
allocation.lease.release()?;
fs::remove_dir_all(self.session_dir(id)).map_err(StandaloneStoreError::Io)?;
fs::remove_dir_all(self.worker_dir(worker_id)).map_err(StandaloneStoreError::Io)?;
sync_directory(&self.root)
}
fn commit_record(
&self,
expected_revision: Option<u64>,
next: &StandaloneSessionRecord,
next: &StandaloneWorkerRecord,
) -> Result<(), StandaloneStoreError> {
let dir = self.session_dir(next.session_id);
let dir = self.worker_dir(next.worker_id);
let marker = dir.join(COMMIT_MARKER);
let mut marker_file = OpenOptions::new()
.write(true)
@@ -417,7 +380,7 @@ impl StandaloneSessionStore {
.open(&marker)
.map_err(|error| {
if error.kind() == io::ErrorKind::AlreadyExists {
StandaloneStoreError::IncompleteCommit(next.session_id)
StandaloneStoreError::IncompleteCommit(next.worker_id)
} else {
StandaloneStoreError::Io(error)
}
@@ -427,11 +390,11 @@ impl StandaloneSessionStore {
sync_directory(&dir)?;
if let Some(expected) = expected_revision {
let current = self.load_record_while_committing(next.session_id)?;
let current = self.load_record_while_committing(next.worker_id)?;
if current.revision != expected {
let _ = fs::remove_file(&marker);
return Err(StandaloneStoreError::RevisionConflict {
id: next.session_id,
id: next.worker_id,
expected,
found: current.revision,
});
@@ -461,30 +424,30 @@ impl StandaloneSessionStore {
fn load_record_while_committing(
&self,
id: StandaloneSessionId,
) -> Result<StandaloneSessionRecord, StandaloneStoreError> {
id: WorkerId,
) -> Result<StandaloneWorkerRecord, StandaloneStoreError> {
let bytes =
fs::read(self.session_dir(id).join(RECORD_FILE)).map_err(StandaloneStoreError::Io)?;
fs::read(self.worker_dir(id).join(RECORD_FILE)).map_err(StandaloneStoreError::Io)?;
serde_json::from_slice(&bytes)
.map_err(|source| StandaloneStoreError::CorruptRecord { id, source })
}
fn session_dir(&self, id: StandaloneSessionId) -> PathBuf {
fn worker_dir(&self, id: WorkerId) -> PathBuf {
self.root.join(id.to_string())
}
}
#[derive(Debug)]
pub struct StandaloneSessionAllocation {
id: StandaloneSessionId,
pub struct StandaloneWorkerAllocation {
worker_id: WorkerId,
cwd: StandaloneCwdIdentity,
lease: StandaloneSessionLease,
lease: StandaloneWorkerLease,
}
impl StandaloneSessionAllocation {
impl StandaloneWorkerAllocation {
#[must_use]
pub fn id(&self) -> StandaloneSessionId {
self.id
pub fn worker_id(&self) -> WorkerId {
self.worker_id
}
#[must_use]
@@ -492,19 +455,19 @@ impl StandaloneSessionAllocation {
&self.cwd
}
pub fn into_lease(self) -> StandaloneSessionLease {
pub fn into_lease(self) -> StandaloneWorkerLease {
self.lease
}
}
#[derive(Debug)]
pub struct StandaloneSessionLease {
pub struct StandaloneWorkerLease {
path: PathBuf,
lease_id: Uuid,
released: bool,
}
impl StandaloneSessionLease {
impl StandaloneWorkerLease {
pub fn release(mut self) -> Result<(), StandaloneStoreError> {
self.release_inner()
}
@@ -534,7 +497,7 @@ impl StandaloneSessionLease {
}
}
impl Drop for StandaloneSessionLease {
impl Drop for StandaloneWorkerLease {
fn drop(&mut self) {
let _ = self.release_inner();
}
@@ -624,7 +587,7 @@ fn classify_lease_liveness(
}
}
fn read_lease(path: &Path, id: StandaloneSessionId) -> Result<LeaseRecord, StandaloneStoreError> {
fn read_lease(path: &Path, id: WorkerId) -> Result<LeaseRecord, StandaloneStoreError> {
let bytes = fs::read(path).map_err(StandaloneStoreError::Io)?;
serde_json::from_slice(&bytes)
.map_err(|source| StandaloneStoreError::CorruptLease { id, source })
@@ -692,47 +655,47 @@ pub enum StandaloneStoreError {
CwdUnavailable(#[source] io::Error),
#[error("standalone cwd is not a directory")]
CwdNotDirectory,
#[error("standalone cwd identity no longer matches the persisted session")]
#[error("standalone cwd identity no longer matches the persisted Worker")]
CwdIdentityMismatch,
#[error("standalone session {0} was not found")]
SessionNotFound(StandaloneSessionId),
#[error("standalone session {0} has an incomplete metadata commit")]
IncompleteCommit(StandaloneSessionId),
#[error("standalone session {0} has invalid metadata")]
InvalidRecord(StandaloneSessionId),
#[error("standalone session {id} metadata is corrupt")]
#[error("standalone Worker {0} was not found")]
WorkerNotFound(WorkerId),
#[error("standalone Worker {0} has an incomplete metadata commit")]
IncompleteCommit(WorkerId),
#[error("standalone Worker {0} has invalid metadata")]
InvalidRecord(WorkerId),
#[error("standalone Worker {id} metadata is corrupt")]
CorruptRecord {
id: StandaloneSessionId,
id: WorkerId,
#[source]
source: serde_json::Error,
},
#[error("standalone session {id} lease is corrupt")]
#[error("standalone Worker {id} lease is corrupt")]
CorruptLease {
id: StandaloneSessionId,
id: WorkerId,
#[source]
source: serde_json::Error,
},
#[error("standalone session {id} uses schema {found}, newer than supported schema {supported}")]
#[error("standalone Worker {id} uses schema {found}, newer than supported schema {supported}")]
NewerSchema {
id: StandaloneSessionId,
id: WorkerId,
found: u32,
supported: u32,
},
#[error("standalone session {0} is already active")]
SessionLeased(StandaloneSessionId),
#[error("standalone session {0} lease liveness cannot be proven; recovery is rejected")]
LeaseLivenessUnknown(StandaloneSessionId),
#[error("standalone session {0} has a stale lease; explicit recovery is required")]
StaleLease(StandaloneSessionId),
#[error("standalone session lease ownership changed")]
#[error("standalone Worker {0} is already active")]
WorkerLeased(WorkerId),
#[error("standalone Worker {0} lease liveness cannot be proven; recovery is rejected")]
LeaseLivenessUnknown(WorkerId),
#[error("standalone Worker {0} has a stale lease; explicit recovery is required")]
StaleLease(WorkerId),
#[error("standalone Worker lease ownership changed")]
LeaseOwnershipLost,
#[error("standalone session {0} must be stopped before deletion")]
DeleteActive(StandaloneSessionId),
#[error("standalone Worker {0} must be stopped before deletion")]
DeleteActive(WorkerId),
#[error(
"standalone session {id} metadata revision changed (expected {expected}, found {found})"
"standalone Worker {id} metadata revision changed (expected {expected}, found {found})"
)]
RevisionConflict {
id: StandaloneSessionId,
id: WorkerId,
expected: u64,
found: u64,
},
crates/standalone/tests/host.rs
+51 -43
View File
@@ -14,7 +14,7 @@ use futures::{Stream, stream};
use protocol::{Event, Method};
use standalone::{
StaleLeasePolicy, StandaloneHost, StandaloneLaunchConfig, StandaloneListScope,
StandaloneSessionStatus, StandaloneSessionStore, StandaloneStartupError, StandaloneStoreError,
StandaloneStartupError, StandaloneStoreError, StandaloneWorkerStatus, StandaloneWorkerStore,
};
use uuid::Uuid;
@@ -90,6 +90,12 @@ async fn in_process_host_runs_text_and_read_tool_then_shuts_down() {
let host = StandaloneHost::start_with_model_client(launch, client)
.await
.expect("start in-process host");
assert_eq!(host.record().worker_name, worker_name);
assert_eq!(host.record().manifest.worker.name, worker_name);
assert_eq!(
host.record().storage_key,
format!("standalone-{}", host.worker_id())
);
let mut protocol_client = host.connect();
protocol_client
@@ -238,7 +244,7 @@ type TestResult = Result<(), Box<dyn std::error::Error>>;
async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope() -> TestResult {
let temp = tempfile::tempdir()?;
let cwd = temp.path().join("project");
let state_dir = temp.path().join("client").join("standalone-sessions");
let state_dir = temp.path().join("client").join("standalone-workers");
std::fs::create_dir_all(&cwd)?;
let launch = StandaloneLaunchConfig::new(
&cwd,
@@ -268,7 +274,7 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope(
],
]);
let host = StandaloneHost::start_with_model_client(launch, first_client).await?;
let session_id = host.session_id();
let worker_id = host.worker_id();
let mut protocol_client = host.connect();
protocol_client
.send(&Method::run_text("first request"))
@@ -283,11 +289,11 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope(
wait_for_run_end(&mut protocol_client).await?;
host.shutdown().await?;
let store = StandaloneSessionStore::open(&state_dir)?;
let store = StandaloneWorkerStore::open(&state_dir)?;
let current = store.list(&cwd, StandaloneListScope::CurrentCwd, 100)?;
assert_eq!(current.len(), 1);
assert_eq!(current[0].session_id, session_id);
assert_eq!(current[0].status, StandaloneSessionStatus::Stopped);
assert_eq!(current[0].worker_id, worker_id);
assert_eq!(current[0].status, StandaloneWorkerStatus::Stopped);
let other_cwd = temp.path().join("other");
std::fs::create_dir(&other_cwd)?;
assert!(
@@ -307,8 +313,13 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope(
]]);
let second_inspection = second_client.clone();
let host =
StandaloneHost::restore_with_model_client(state_dir.clone(), session_id, second_client)
StandaloneHost::restore_with_model_client(state_dir.clone(), worker_id, second_client)
.await?;
assert_eq!(
host.record().worker_name,
"display-name-is-not-session-identity"
);
assert_eq!(host.record().storage_key, format!("standalone-{worker_id}"));
let mut protocol_client = host.connect();
let snapshot = format!(
"{:?}",
@@ -338,11 +349,11 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope(
assert!(projected.contains("persisted task"), "{projected}");
host.shutdown().await?;
store.delete(session_id)?;
store.delete(worker_id)?;
assert!(cwd.exists(), "deleting session state must not mutate cwd");
assert!(matches!(
store.load(session_id),
Err(StandaloneStoreError::SessionNotFound(_))
store.load(worker_id),
Err(StandaloneStoreError::WorkerNotFound(_))
));
Ok(())
}
@@ -363,28 +374,25 @@ async fn standalone_restore_rejects_concurrent_lease_and_missing_cwd() -> TestRe
.resolve()?;
let host =
StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?;
let session_id = host.session_id();
let store = StandaloneSessionStore::open(&state_dir)?;
let worker_id = host.worker_id();
let store = StandaloneWorkerStore::open(&state_dir)?;
assert!(matches!(
store.acquire_lease(session_id, StaleLeasePolicy::Recover),
Err(StandaloneStoreError::SessionLeased(id)) if id == session_id
store.acquire_lease(worker_id, StaleLeasePolicy::Recover),
Err(StandaloneStoreError::WorkerLeased(id)) if id == worker_id
));
let restore = StandaloneHost::restore_with_model_client(
state_dir.clone(),
session_id,
worker_id,
ScriptedClient::new(Vec::new()),
)
.await;
assert!(matches!(
restore,
Err(StandaloneStartupError::SessionActive)
));
assert!(matches!(restore, Err(StandaloneStartupError::WorkerActive)));
host.shutdown().await?;
std::fs::rename(&cwd, &moved)?;
let restore = StandaloneHost::restore_with_model_client(
state_dir,
session_id,
worker_id,
ScriptedClient::new(Vec::new()),
)
.await;
@@ -421,11 +429,11 @@ async fn standalone_restore_recovers_only_a_proven_stale_lease() -> TestResult {
});
let host =
StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?;
let session_id = host.session_id();
let worker_id = host.worker_id();
host.shutdown().await?;
let store = StandaloneSessionStore::open(&state_dir)?;
let store = StandaloneWorkerStore::open(&state_dir)?;
assert!(matches!(
store.load(session_id)?.manifest.profile,
store.load(worker_id)?.manifest.profile,
Some(manifest::ProfileManifestSnapshot {
source: manifest::ProfileSource::Registry {
source: manifest::ProfileRegistrySource::User,
@@ -434,9 +442,9 @@ async fn standalone_restore_recovers_only_a_proven_stale_lease() -> TestResult {
..
})
));
let session_dir = state_dir.join(session_id.to_string());
let worker_dir = state_dir.join(worker_id.to_string());
std::fs::write(
session_dir.join("lease.json"),
worker_dir.join("lease.json"),
serde_json::to_vec(&serde_json::json!({
"lease_id": uuid::Uuid::now_v7(),
"pid": u32::MAX,
@@ -447,7 +455,7 @@ async fn standalone_restore_recovers_only_a_proven_stale_lease() -> TestResult {
let host = StandaloneHost::restore_with_model_client(
state_dir,
session_id,
worker_id,
ScriptedClient::new(Vec::new()),
)
.await?;
@@ -468,11 +476,11 @@ async fn standalone_restore_rejects_lease_with_missing_start_marker() -> TestRes
.resolve()?;
let host =
StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?;
let session_id = host.session_id();
let worker_id = host.worker_id();
host.shutdown().await?;
let session_dir = state_dir.join(session_id.to_string());
let worker_dir = state_dir.join(worker_id.to_string());
std::fs::write(
session_dir.join("lease.json"),
worker_dir.join("lease.json"),
serde_json::to_vec(&serde_json::json!({
"lease_id": uuid::Uuid::now_v7(),
"pid": std::process::id(),
@@ -480,14 +488,14 @@ async fn standalone_restore_rejects_lease_with_missing_start_marker() -> TestRes
}))?,
)?;
let store = StandaloneSessionStore::open(&state_dir)?;
let store = StandaloneWorkerStore::open(&state_dir)?;
assert!(matches!(
store.acquire_lease(session_id, StaleLeasePolicy::Recover),
Err(StandaloneStoreError::LeaseLivenessUnknown(id)) if id == session_id
store.acquire_lease(worker_id, StaleLeasePolicy::Recover),
Err(StandaloneStoreError::LeaseLivenessUnknown(id)) if id == worker_id
));
let restore = StandaloneHost::restore_with_model_client(
state_dir,
session_id,
worker_id,
ScriptedClient::new(Vec::new()),
)
.await;
@@ -511,23 +519,23 @@ async fn standalone_metadata_fails_closed_on_incomplete_or_newer_records() -> Te
.resolve()?;
let host =
StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?;
let session_id = host.session_id();
let worker_id = host.worker_id();
host.shutdown().await?;
let store = StandaloneSessionStore::open(&state_dir)?;
let session_dir = state_dir.join(session_id.to_string());
std::fs::write(session_dir.join("commit.pending"), b"interrupted\n")?;
let store = StandaloneWorkerStore::open(&state_dir)?;
let worker_dir = state_dir.join(worker_id.to_string());
std::fs::write(worker_dir.join("commit.pending"), b"interrupted\n")?;
assert!(matches!(
store.load(session_id),
Err(StandaloneStoreError::IncompleteCommit(id)) if id == session_id
store.load(worker_id),
Err(StandaloneStoreError::IncompleteCommit(id)) if id == worker_id
));
std::fs::remove_file(session_dir.join("commit.pending"))?;
let record_path = session_dir.join("record.json");
std::fs::remove_file(worker_dir.join("commit.pending"))?;
let record_path = worker_dir.join("record.json");
let mut record: serde_json::Value = serde_json::from_slice(&std::fs::read(&record_path)?)?;
record["schema_version"] = serde_json::json!(u32::MAX);
std::fs::write(&record_path, serde_json::to_vec_pretty(&record)?)?;
assert!(matches!(
store.load(session_id),
Err(StandaloneStoreError::NewerSchema { id, .. }) if id == session_id
store.load(worker_id),
Err(StandaloneStoreError::NewerSchema { id, .. }) if id == worker_id
));
Ok(())
}