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

feat: merge develop into hare/develop

Browse Source
This commit is contained in:
Hare
2026-09-21 04:30:02 +09:00
parent 5503b9b977 0df9913286
commit 14d21c4642
77 changed files with 8421 additions and 2482 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/hosts.rs
+162 -41
View File
@@ -32,8 +32,9 @@ use worker_runtime::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource,
RepositoryRefObservation, RepositoryRefObservationRequest,
WorkerDetail as EmbeddedWorkerDetail, WorkerStatus as EmbeddedWorkerStatus,
WorkingDirectoryClaim, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef,
WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
WorkingDirectoryAttachmentStatus, WorkingDirectoryRepositoryAccessRequest,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkspaceApiRef,
};
use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
#[cfg(test)]
@@ -270,8 +271,8 @@ pub struct WorkerSummary {
pub retention_state: String,
pub implementation: WorkerImplementationSummary,
pub capabilities: WorkerCapabilitySummary,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectorySummary>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
@@ -336,7 +337,7 @@ impl From<RuntimeSummary> for server_api::RuntimeSummary {
pub(crate) fn workspace_worker_summary(
summary: WorkerSummary,
resource_key: String,
working_directory: Option<server_api::WorkingDirectorySummary>,
workdir_attachments: Vec<server_api::WorkerWorkdirAttachmentSummary>,
) -> server_api::WorkerSummary {
server_api::WorkerSummary {
runtime_id: summary.worker.runtime_id,
@@ -366,7 +367,7 @@ pub(crate) fn workspace_worker_summary(
can_stop: summary.capabilities.can_stop,
can_spawn_followup: summary.capabilities.can_spawn_followup,
},
working_directory,
workdir_attachments,
diagnostics: summary.diagnostics.into_iter().map(Into::into).collect(),
}
}
@@ -483,6 +484,13 @@ pub struct WorkerCreateBinding {
pub create_fingerprint: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WorkerSpawnWorkingDirectoryAttachmentRequest {
pub alias: workdir::WorkdirAttachmentAlias,
pub working_directory: WorkerSpawnWorkingDirectoryRequest,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WorkerSpawnRequest {
@@ -495,15 +503,13 @@ pub struct WorkerSpawnRequest {
pub ticket_assignment: Option<WorkerTicketAssignmentRequest>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub initial_submit: Vec<Segment>,
/// Optional safe working-directory creation request. The Workspace server resolves
/// this into a runtime-internal `WorkingDirectoryRequest` from configured
/// repositories before calling a host.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory_request: Option<WorkerSpawnWorkingDirectoryRequest>,
/// Safe working-directory materialization requests, keyed by stable Worker-local aliases.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachment_requests: Vec<WorkerSpawnWorkingDirectoryAttachmentRequest>,
#[serde(skip, default)]
pub resolved_working_directory_request: Option<WorkingDirectoryRequest>,
pub resolved_workdir_attachment_requests: Vec<WorkingDirectoryAttachmentRequest>,
#[serde(skip, default)]
pub resolved_working_directory: Option<WorkingDirectoryClaim>,
pub resolved_workdir_attachments: Vec<WorkingDirectoryAttachmentClaim>,
#[serde(skip, default)]
pub resolved_config_bundle: Option<ConfigBundle>,
#[serde(skip, default)]
@@ -905,6 +911,24 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
}
}
fn replace_worker_workdir_attachments(
&self,
worker_id: &str,
_attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> WorkerWorkspaceApiResult {
WorkerWorkspaceApiResult {
state: WorkerOperationState::Unsupported,
worker: None,
diagnostics: vec![diagnostic(
"worker_workdir_attachments_replace_unsupported",
DiagnosticSeverity::Info,
format!(
"runtime does not support replacing Workdir attachments for worker `{worker_id}`"
),
)],
}
}
fn create_working_directory(
&self,
_request: WorkingDirectoryRequest,
@@ -1489,6 +1513,19 @@ impl RuntimeRegistry {
Ok(runtime.replace_worker_workspace_api(worker_id, workspace_api))
}
pub fn replace_worker_workdir_attachments(
&self,
worker: &RuntimeWorkerRef,
attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> Result<WorkerWorkspaceApiResult, RuntimeRegistryError> {
let runtime_id = worker.runtime_id.as_str();
let worker_id = worker.worker_id.as_str();
validate_backend_identifier("runtime_id", runtime_id)?;
validate_backend_identifier("worker_id", worker_id)?;
let runtime = self.runtime(runtime_id)?;
Ok(runtime.replace_worker_workdir_attachments(worker_id, attachments))
}
pub fn observe_workspace_prompt_projection(
&self,
projection: worker::WorkspacePromptProjection,
@@ -2176,7 +2213,7 @@ impl EmbeddedWorkerRuntime {
&& self.can_stop_embedded_worker(summary.status),
can_spawn_followup: false,
},
working_directory: summary.working_directory.map(|status| status.summary),
workdir_attachments: summary.workdir_attachments,
diagnostics: embedded_worker_projection_diagnostics(
summary.execution_metadata_available,
),
@@ -2220,7 +2257,7 @@ impl EmbeddedWorkerRuntime {
&& self.can_stop_embedded_worker(detail.status),
can_spawn_followup: false,
},
working_directory: detail.working_directory.map(|status| status.summary),
workdir_attachments: detail.workdir_attachments,
diagnostics: embedded_worker_projection_diagnostics(
detail.execution_metadata_available,
),
@@ -2429,6 +2466,39 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
}
}
fn replace_worker_workdir_attachments(
&self,
worker_id: &str,
attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> WorkerWorkspaceApiResult {
let Some(worker_ref) = self.worker_ref(worker_id) else {
return WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic(
"embedded_worker_id_invalid",
DiagnosticSeverity::Warning,
"Worker id was empty and cannot change Workdir attachments".to_string(),
)],
};
};
match self
.runtime
.replace_worker_workdir_attachments(&worker_ref, attachments)
{
Ok(detail) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(detail)),
diagnostics: Vec::new(),
},
Err(err) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![embedded_runtime_diagnostic(&err)],
},
}
}
fn create_working_directory(
&self,
_request: WorkingDirectoryRequest,
@@ -2491,8 +2561,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
request: WorkerSpawnRequest,
) -> WorkerSpawnResult {
let mut diagnostics = Vec::new();
if request.resolved_working_directory_request.is_some()
|| request.resolved_working_directory.is_some()
if !request.resolved_workdir_attachment_requests.is_empty()
|| !request.resolved_workdir_attachments.is_empty()
{
diagnostics.push(embedded_workdir_unsupported_diagnostic());
return WorkerSpawnResult {
@@ -2573,8 +2643,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
config_bundle,
profile_source,
initial_input: initial_worker_input(&request.initial_submit),
working_directory_request: request.resolved_working_directory_request.clone(),
working_directory: request.resolved_working_directory.clone(),
workdir_attachment_requests: request.resolved_workdir_attachment_requests.clone(),
workdir_attachments: request.resolved_workdir_attachments.clone(),
worker_observation_enabled: request.resolved_worker_observation_enabled,
worker_observation_grants: request.resolved_worker_observation_grants.clone(),
workspace_api: Some(workspace_api),
@@ -3982,7 +4052,7 @@ impl RemoteWorkerRuntime {
&& runtime_worker_can_stop(true, summary.status),
can_spawn_followup: false,
},
working_directory: summary.working_directory.map(|status| status.summary),
workdir_attachments: summary.workdir_attachments,
diagnostics: remote_worker_projection_diagnostics(summary.execution_metadata_available),
}
}
@@ -4024,7 +4094,7 @@ impl RemoteWorkerRuntime {
&& runtime_worker_can_stop(true, detail.status),
can_spawn_followup: false,
},
working_directory: detail.working_directory.map(|status| status.summary),
workdir_attachments: detail.workdir_attachments,
diagnostics: remote_worker_projection_diagnostics(detail.execution_metadata_available),
}
}
@@ -4388,6 +4458,51 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
}
}
fn replace_worker_workdir_attachments(
&self,
worker_id: &str,
attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> WorkerWorkspaceApiResult {
let workdir_attachments = match runtime_contract_convert(attachments) {
Ok(attachments) => attachments,
Err(diagnostic) => {
return WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic],
};
}
};
let request = runtime_api::WorkerWorkdirAttachmentsRequest {
workdir_attachments,
};
let worker_id = worker_id.to_string();
match self
.run_runtime_api(
self.request_timeout,
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move {
client
.replace_worker_workdir_attachments(worker_id, request)
.await
},
)
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker)
}) {
Ok(response) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(response)),
diagnostics: Vec::new(),
},
Err(diagnostic) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic],
},
}
}
fn create_working_directory(
&self,
request: WorkingDirectoryRequest,
@@ -4567,8 +4682,8 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
config_bundle,
profile_source,
initial_input: initial_worker_input(&request.initial_submit),
working_directory_request: request.resolved_working_directory_request.clone(),
working_directory: request.resolved_working_directory.clone(),
workdir_attachment_requests: request.resolved_workdir_attachment_requests.clone(),
workdir_attachments: request.resolved_workdir_attachments.clone(),
worker_observation_enabled: request.resolved_worker_observation_enabled,
worker_observation_grants: request.resolved_worker_observation_grants.clone(),
workspace_api: Some(workspace_api),
@@ -5724,7 +5839,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> WorkerSummary {
can_stop: false,
can_spawn_followup: false,
},
working_directory: None,
workdir_attachments: Vec::new(),
diagnostics: vec![diagnostic(
"runtime_capability_unsupported",
DiagnosticSeverity::Info,
@@ -6265,10 +6380,16 @@ mod tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
workdir_attachments: request
.workdir_attachments
.iter()
.map(|(alias, binding)| {
worker_runtime::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
}
})
.collect(),
}
}
@@ -6360,7 +6481,7 @@ mod tests {
can_stop: false,
can_spawn_followup: false,
},
working_directory: None,
workdir_attachments: Vec::new(),
diagnostics: Vec::new(),
}],
observed_prompt_revisions: Arc::new(Mutex::new(Vec::new())),
@@ -6650,9 +6771,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
@@ -6890,9 +7011,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
@@ -6990,9 +7111,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
@@ -7029,9 +7150,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:companion".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
crates/workspace-server/src/latest_schema.sql
+5 -4
View File
@@ -667,6 +667,7 @@ CREATE TABLE workdir_create_credential_revision_retentions (
CREATE TABLE "workdir_registry" (
workspace_id TEXT NOT NULL,
workdir_id TEXT NOT NULL,
display_name TEXT,
runtime_id TEXT NOT NULL,
repository_id TEXT NOT NULL,
creation_selector TEXT,
@@ -859,10 +860,10 @@ CREATE TABLE "worker_workdir_links" (
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
workdir_id TEXT NOT NULL,
role TEXT NOT NULL,
alias TEXT NOT NULL,
linked_at TEXT NOT NULL,
unlinked_at TEXT,
PRIMARY KEY (workspace_id, worker_id, workdir_id, role),
PRIMARY KEY (workspace_id, worker_id, workdir_id, alias),
FOREIGN KEY (workspace_id, worker_id)
REFERENCES "worker_registry"(workspace_id, worker_id) ON DELETE CASCADE,
FOREIGN KEY (workspace_id, workdir_id)
@@ -1093,8 +1094,8 @@ CREATE INDEX worker_removal_operations_worker_idx ON worker_removal_operations(w
CREATE UNIQUE INDEX worker_workdir_links_active_workdir_unique
ON worker_workdir_links(workspace_id, workdir_id)
WHERE unlinked_at IS NULL;
CREATE UNIQUE INDEX worker_workdir_links_active_worker_unique
ON worker_workdir_links(workspace_id, worker_id)
CREATE UNIQUE INDEX worker_workdir_links_active_alias_unique
ON worker_workdir_links(workspace_id, worker_id, alias)
WHERE unlinked_at IS NULL;
CREATE INDEX worker_workdir_links_workdir
ON worker_workdir_links(workspace_id, workdir_id);
crates/workspace-server/src/retention.rs
+132 -5
View File
@@ -2,7 +2,10 @@
//! Runtime receives only resolved dispositions and stable ids; policy authority
//! never comes from prompts, profiles, or model input.
use crate::{Error as StoreError, store::SqliteWorkspaceStore};
use crate::{
Error as StoreError,
store::{SqliteWorkspaceStore, WorkerRemovalCommit, commit_worker_catalog_removal},
};
use chrono::Utc;
use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
use serde::{Deserialize, Serialize};
@@ -417,13 +420,19 @@ impl SqliteWorkspaceStore {
operation_id: &str,
fp: &str,
result: &WorkerRetentionExecutionResult,
) -> Result<WorkerRemovalPlan, WorkerRetentionError> {
) -> Result<WorkerRemovalCommit, WorkerRetentionError> {
self.with_conn_mut(|conn|{
let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let mut plan=load_plan_op(&tx,operation_id)?.ok_or_else(||StoreError::InvalidInput("operation missing".into()))?;
if plan.workspace_id!=workspace_id{return Err(StoreError::InvalidInput("cross-workspace".into()));}
if plan.input_fingerprint!=fp||result.input_fingerprint!=fp||result.operation_id!=operation_id{return Err(StoreError::InvalidInput(format!("fingerprint:{operation_id}")));}
if plan.state==WorkerRemovalPlanState::Succeeded{tx.commit()?;return Ok(plan);}
if plan.state==WorkerRemovalPlanState::Succeeded{
// Repair projection fences left by the pre-atomic implementation and
// return a replayable change for a broadcast missed after commit.
let catalog=commit_worker_catalog_removal(&tx,workspace_id,&plan.worker)?;
tx.commit()?;
return Ok(WorkerRemovalCommit { plan, catalog });
}
if plan.state != WorkerRemovalPlanState::Executing {
return Err(StoreError::InvalidInput(format!("stale:{}:plan state {} is not committable", plan.plan_id, state_s(plan.state))));
}
@@ -460,12 +469,14 @@ impl SqliteWorkspaceStore {
if plan.metadata_disposition==MetadataDisposition::Tombstone{
tx.execute("INSERT OR IGNORE INTO worker_tombstones(workspace_id,runtime_id,worker_id,display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_revision,operation_id) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,worker.display_name,worker.profile,worker.created_at,now,plan.archive_id,plan.policy_id,plan.policy_revision,operation_id])?;
}
tx.execute("UPDATE worker_workdir_links SET unlinked_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND unlinked_at IS NULL",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?;
let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND updated_at=?4",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_revision])?;
tx.execute("UPDATE worker_create_reservations SET state='removed',updated_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state='created'",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?;
if deleted!=1{return Err(StoreError::InvalidInput(format!("stale:{}:removal fence changed",plan.plan_id)));}
let catalog=commit_worker_catalog_removal(&tx,workspace_id,&plan.worker)?;
tx.execute("UPDATE worker_removal_operations SET state='succeeded',failure_category=NULL,updated_at=?1 WHERE operation_id=?2",params![now,operation_id])?;
tx.execute("INSERT OR IGNORE INTO worker_retention_audit_events(event_id,operation_id,workspace_id,event_kind,detail,created_at) VALUES(?1,?2,?3,'worker_removed',?4,?5)",params![stable("wre",operation_id),operation_id,workspace_id,format!("runtime_id={} worker_id={} session={} metadata={} diagnostics={}",plan.worker.runtime_id,plan.worker.worker_id,sess(plan.session_disposition),meta(plan.metadata_disposition),diag(plan.diagnostics_disposition)),now])?;
tx.commit()?;plan.state=WorkerRemovalPlanState::Succeeded;plan.updated_at=now;Ok(plan)
tx.commit()?;plan.state=WorkerRemovalPlanState::Succeeded;plan.updated_at=now;Ok(WorkerRemovalCommit { plan, catalog })
}).map_err(map_error)
}
@@ -948,7 +959,9 @@ fn parse_state(v: &str) -> rusqlite::Result<WorkerRemovalPlanState> {
#[cfg(test)]
mod tests {
use super::*;
use crate::store::{ControlPlaneStore, TicketCoderAssignmentRecord, WorkerRegistryRecord};
use crate::store::{
ControlPlaneStore, TicketCoderAssignmentRecord, WorkerCatalogChange, WorkerRegistryRecord,
};
use worker_runtime::identity::WorkerId;
fn worker_id() -> WorkerId {
WorkerId::from_legacy_u64(1)
@@ -1203,6 +1216,7 @@ mod tests {
assert_eq!(
s.commit_worker_removal("w", &p.operation_id, &p.input_fingerprint, &result)
.unwrap()
.plan
.state,
WorkerRemovalPlanState::Succeeded
);
@@ -1219,6 +1233,7 @@ mod tests {
assert_eq!(
s.commit_worker_removal("w", &p.operation_id, &p.input_fingerprint, &result)
.unwrap()
.plan
.state,
WorkerRemovalPlanState::Succeeded
);
@@ -1351,6 +1366,98 @@ mod tests {
.unwrap();
assert!(s.worker_tombstone("w", &p.worker).unwrap().is_none());
}
#[test]
fn terminal_catalog_failure_rolls_back_removal_projection_and_operation() {
let store = setup();
let purge = WorkerRetentionPolicyUpdate {
policy_id: "purge".into(),
session_disposition: SessionDisposition::Purge,
metadata_disposition: MetadataDisposition::Tombstone,
archive_retention: ArchiveRetention::Forever,
diagnostics_disposition: DiagnosticsDisposition::Purge,
diagnostics_retention_seconds: None,
};
store
.update_worker_retention_policy("w", 1, &purge)
.unwrap();
let plan = store.plan_worker_removal(&req(), &inv()).unwrap();
store
.prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint)
.unwrap();
let result = WorkerRetentionExecutionResult {
operation_id: plan.operation_id.clone(),
input_fingerprint: plan.input_fingerprint.clone(),
expected_worker_revision: plan.worker_revision.clone(),
worker_id: worker_id(),
session_disposition: SessionDisposition::Purge,
diagnostics_disposition: DiagnosticsDisposition::Purge,
archive: None,
source_removed: true,
diagnostics_retained: false,
};
store
.with_conn(|conn| {
conn.execute_batch(
"CREATE TRIGGER fail_worker_projection_removal
BEFORE INSERT ON worker_registry_projection_removals
BEGIN SELECT RAISE(ABORT, 'injected projection failure'); END;",
)?;
Ok(())
})
.unwrap();
assert!(
store
.commit_worker_removal("w", &plan.operation_id, &plan.input_fingerprint, &result,)
.is_err()
);
let (registry, removal, state, tombstone): (i64, i64, String, i64) = store
.with_conn(|conn| {
Ok((
conn.query_row(
"SELECT COUNT(*) FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1",
[worker_id().to_string()],
|row| row.get(0),
)?,
conn.query_row(
"SELECT COUNT(*) FROM worker_registry_projection_removals WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1",
[worker_id().to_string()],
|row| row.get(0),
)?,
conn.query_row(
"SELECT state FROM worker_removal_operations WHERE operation_id=?1",
[&plan.operation_id],
|row| row.get(0),
)?,
conn.query_row(
"SELECT COUNT(*) FROM worker_tombstones WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1",
[worker_id().to_string()],
|row| row.get(0),
)?,
))
})
.unwrap();
assert_eq!(registry, 1, "catalog deletion must roll back");
assert_eq!(removal, 0, "removal fence must not partially commit");
assert_eq!(state, "executing", "operation must remain recoverable");
assert_eq!(tombstone, 0, "retention metadata must roll back");
store
.with_conn(|conn| {
conn.execute_batch("DROP TRIGGER fail_worker_projection_removal;")?;
Ok(())
})
.unwrap();
let committed = store
.commit_worker_removal("w", &plan.operation_id, &plan.input_fingerprint, &result)
.unwrap();
assert!(matches!(
committed.catalog.changes.as_slice(),
[WorkerCatalogChange::Removed(worker)] if worker == &plan.worker
));
assert_eq!(committed.plan.state, WorkerRemovalPlanState::Succeeded);
}
#[test]
fn commit_requires_executing_state_and_exact_runtime_result() {
let store = setup();
@@ -1597,6 +1704,26 @@ mod tests {
.unwrap()
.is_none()
);
s.with_conn(|conn| {
conn.execute(
"DELETE FROM worker_registry_projection_removals WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1",
[worker_id().to_string()],
)?;
Ok(())
})
.unwrap();
let repaired = s
.recover_succeeded_worker_removal_catalog("w", &request.worker)
.unwrap();
assert!(matches!(
repaired.changes.as_slice(),
[WorkerCatalogChange::Removed(worker)] if worker == &request.worker
));
assert!(
s.get_worker_registry("w", &request.worker)
.unwrap()
.is_none()
);
let recovered = s
.recover_worker_removal_execution("w", &request.worker)
.unwrap()
crates/workspace-server/src/runtime_subscription_tests.rs
+3 -3
View File
@@ -22,7 +22,7 @@ impl WorkerExecutionBackend for TestExecutionBackend {
protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
None,
Vec::new(),
)
}
@@ -64,8 +64,8 @@ fn create_request(name: &str) -> CreateWorkerRequest {
.unwrap(),
},
initial_input: None,
working_directory_request: None,
working_directory: None,
workdir_attachment_requests: Vec::new(),
workdir_attachments: Vec::new(),
worker_observation_enabled: false,
worker_observation_grants: Vec::new(),
workspace_api: None,
crates/workspace-server/src/server.rs
+2101 -917
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/store.rs
+394 -184
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/workdir_create_operations.rs
+4 -1
View File
@@ -25,6 +25,7 @@ pub fn request_fingerprint(
repository_id: &str,
selector: Option<&str>,
requested_runtime_id: Option<&str>,
display_name: Option<&str>,
repository_source_fingerprint: &str,
repository_source_revision: u64,
) -> String {
@@ -33,6 +34,7 @@ pub fn request_fingerprint(
Some(repository_id),
selector,
requested_runtime_id,
display_name,
Some(repository_source_fingerprint),
] {
match value {
@@ -569,6 +571,7 @@ mod tests {
"main",
Some("develop"),
None,
None,
"sha256:test",
1,
),
@@ -718,7 +721,7 @@ mod tests {
);
let mut changed_input = record.clone();
changed_input.request_fingerprint =
request_fingerprint("main", Some("main"), None, "sha256:test", 1);
request_fingerprint("main", Some("main"), None, None, "sha256:test", 1);
assert!(
store
.reserve_workdir_create_operation(&changed_input)
crates/workspace-server/src/workdir_removal.rs
+16 -14
View File
@@ -716,7 +716,7 @@ fn load_workdir_record(
workdir_id: &str,
) -> Result<Option<WorkdirRegistryRecord>> {
conn.query_row(
r#"SELECT workspace_id, workdir_id, runtime_id, repository_id,
r#"SELECT workspace_id, workdir_id, display_name, runtime_id, repository_id,
creation_selector, creation_ref, creation_tree,
current_selector, current_ref, current_tree, observed_at_epoch_seconds,
materialization_status, cleanliness, created_at, updated_at
@@ -726,19 +726,20 @@ fn load_workdir_record(
Ok(WorkdirRegistryRecord {
workspace_id: row.get(0)?,
workdir_id: row.get(1)?,
runtime_id: row.get(2)?,
repository_id: row.get(3)?,
creation_selector: row.get(4)?,
creation_ref: row.get(5)?,
creation_tree: row.get(6)?,
current_selector: row.get(7)?,
current_ref: row.get(8)?,
current_tree: row.get(9)?,
observed_at_epoch_seconds: row.get::<_, Option<i64>>(10)?.map(|value| value as u64),
materialization_status: row.get(11)?,
cleanliness: row.get(12)?,
created_at: row.get(13)?,
updated_at: row.get(14)?,
display_name: row.get(2)?,
runtime_id: row.get(3)?,
repository_id: row.get(4)?,
creation_selector: row.get(5)?,
creation_ref: row.get(6)?,
creation_tree: row.get(7)?,
current_selector: row.get(8)?,
current_ref: row.get(9)?,
current_tree: row.get(10)?,
observed_at_epoch_seconds: row.get::<_, Option<i64>>(11)?.map(|value| value as u64),
materialization_status: row.get(12)?,
cleanliness: row.get(13)?,
created_at: row.get(14)?,
updated_at: row.get(15)?,
})
},
)
@@ -934,6 +935,7 @@ mod tests {
let workdir = WorkdirRegistryRecord {
workspace_id: "workspace-a".to_string(),
workdir_id: "workdir-a".to_string(),
display_name: None,
runtime_id: "runtime-a".to_string(),
repository_id: "repository-a".to_string(),
creation_selector: Some("refs/heads/develop".to_string()),
crates/workspace-server/src/worker_projection.rs
+17 -54
View File
@@ -11,18 +11,12 @@ use tokio::task::JoinHandle;
use crate::runtime_subscription::{
BrokerSubscriptionEvent, RuntimeSubscriptionBroker, RuntimeSubscriptionBrokerError,
};
use crate::store::{
ControlPlaneStore, WorkerRegistryProjectionCommit, WorkerRegistryProjectionRecord,
};
use crate::store::{ControlPlaneStore, WorkerCatalogChange, WorkerRegistryProjectionCommit};
use worker_runtime::identity::RuntimeWorkerRef;
const PROJECTION_EVENT_CAPACITY: usize = 256;
#[derive(Debug, Clone)]
pub enum WorkerProjectionChange {
Upsert(WorkerRegistryProjectionRecord),
Removed(RuntimeWorkerRef),
}
pub type WorkerProjectionChange = WorkerCatalogChange;
#[derive(Debug, Clone)]
pub struct WorkerProjectionEvent {
@@ -144,27 +138,13 @@ impl WorkerProjectionService {
self.publish_commit(commit)
}
pub fn publish_catalog_change(&self, worker: &RuntimeWorkerRef) -> crate::Result<()> {
pub fn refresh(&self, worker: &RuntimeWorkerRef) -> crate::Result<()> {
let commit = self
.store
.publish_worker_registry_catalog_change(self.workspace_id.as_str(), worker)?;
.commit_worker_registry_projection_refresh(self.workspace_id.as_str(), worker)?;
self.publish_commit(commit)
}
pub fn publish_removed(&self, worker: RuntimeWorkerRef) -> crate::Result<()> {
let commit = self
.store
.publish_worker_registry_removal(self.workspace_id.as_str(), &worker)?;
if commit.changed_workers.is_empty() {
return Ok(());
}
let _ = self.events.send(WorkerProjectionEvent {
revision: commit.revision,
changes: vec![WorkerProjectionChange::Removed(worker)],
});
Ok(())
}
async fn apply_broker_event(
&self,
runtime_id: &str,
@@ -236,38 +216,21 @@ impl WorkerProjectionService {
self.publish_commit(commit)
}
fn publish_commit(&self, commit: WorkerRegistryProjectionCommit) -> crate::Result<()> {
if commit.changed_workers.is_empty() {
/// Broadcast an already committed catalog payload. Persistence always happens
/// before this call; a process crash therefore closes subscribers, whose next
/// subscription starts from the durable snapshot. Retrying a removal returns
/// the same durable change and can heal an in-process missed broadcast.
pub(crate) fn publish_commit(
&self,
commit: WorkerRegistryProjectionCommit,
) -> crate::Result<()> {
if commit.changes.is_empty() {
return Ok(());
}
let changes = commit
.changed_workers
.iter()
.filter_map(|worker| {
match self
.store
.worker_registry_projection(self.workspace_id.as_str(), worker)
{
Ok(Some(record)) => Some(WorkerProjectionChange::Upsert(record)),
Ok(None) => None,
Err(error) => {
tracing::warn!(
runtime_id = %worker.runtime_id,
worker_id = %worker.worker_id,
error = %error,
"failed to load committed Worker projection"
);
None
}
}
})
.collect::<Vec<_>>();
if !changes.is_empty() {
let _ = self.events.send(WorkerProjectionEvent {
revision: commit.revision,
changes,
});
}
let _ = self.events.send(WorkerProjectionEvent {
revision: commit.revision,
changes: commit.changes,
});
Ok(())
}
}
crates/workspace-server/src/workspace_deletion.rs
+3 -1
View File
@@ -1198,7 +1198,7 @@ mod tests {
);
assert!(
store
!store
.delete_worker_registry(
&workspace_id,
&worker_runtime::identity::RuntimeWorkerRef {
@@ -1207,6 +1207,8 @@ mod tests {
},
)
.expect("remove worker registry")
.changes
.is_empty()
);
let state: String = store
.with_conn(|conn| {
crates/workspace-server/src/workspace_subscription.rs
+43 -33
View File
@@ -413,9 +413,7 @@ fn project_registry_worker(
workspace_id: Some(record.registry.workspace_id.clone()),
display_name: Some(record.registry.display_name.clone()),
profile: record.registry.profile.clone(),
repository_id: None,
repository_key: None,
working_directory_id: None,
workdir_attachments: Vec::new(),
}
};
worker.runtime_id = Some(runtime_id);
@@ -440,20 +438,30 @@ fn project_working_directory(
else {
return false;
};
if let Some(link) = links.first() {
let Ok(Some(workdir)) = api
.store
.get_workdir_registry(&api.config.workspace_id, &link.workdir_id)
else {
return false;
};
let Ok(working_directory_id) =
protocol::subscription::SubscriptionWorkdirId::new(workdir.workdir_id)
else {
return false;
};
worker.repository_id = Some(workdir.repository_id);
worker.working_directory_id = Some(working_directory_id);
if !links.is_empty() {
let mut attachments = Vec::with_capacity(links.len());
for link in links {
let Ok(Some(workdir)) = api
.store
.get_workdir_registry(&api.config.workspace_id, &link.workdir_id)
else {
return false;
};
let Ok(working_directory_id) =
protocol::subscription::SubscriptionWorkdirId::new(workdir.workdir_id)
else {
return false;
};
attachments.push(
protocol::subscription::SubscriptionWorkerWorkdirAttachment {
alias: link.alias,
repository_id: Some(workdir.repository_id),
repository_key: None,
working_directory_id,
},
);
}
worker.workdir_attachments = attachments;
} else {
let Ok(has_link_history) = api
.store
@@ -462,26 +470,28 @@ fn project_working_directory(
return false;
};
if has_link_history {
worker.repository_id = None;
worker.repository_key = None;
worker.working_directory_id = None;
worker.workdir_attachments.clear();
}
}
project_repository_key(api, worker)
project_repository_keys(api, worker)
}
fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
let Some(repository_id) = worker.repository_id.take() else {
worker.repository_key = None;
return true;
};
let Ok(Some(repository)) = api
.store
.get_repository(&api.config.workspace_id, &repository_id)
else {
return false;
};
worker.repository_key = Some(repository.repository_key);
fn project_repository_keys(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
for attachment in &mut worker.workdir_attachments {
let Some(repository_id) = attachment.repository_id.take() else {
if attachment.repository_key.is_none() {
return false;
}
continue;
};
let Ok(Some(repository)) = api
.store
.get_repository(&api.config.workspace_id, &repository_id)
else {
return false;
};
attachment.repository_key = Some(repository.repository_key);
}
true
}