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

fix: preserve repository access across retries

Browse Source
This commit is contained in:
Hare
2026-08-26 15:21:52 +09:00
parent ffb2a34ae5
commit 471db64bcc
5 changed files with 570 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/repository_access.rs
+72 -20
View File
@@ -865,33 +865,73 @@ impl RepositorySecretService {
binding.host_trust_id
))
})?;
let (private_key, passphrase) = self.store.with_conn(|conn| {
self.lease_ssh_materialization_access_revision(
workspace_id,
&binding.credential_id,
credential.current_revision,
&binding.host_trust_id,
host_trust.current_revision,
)
}
pub fn lease_ssh_materialization_access_revision(
&self,
workspace_id: &str,
credential_id: &str,
credential_revision: u64,
host_trust_id: &str,
host_trust_revision: u64,
) -> Result<LeasedRepositorySshAccess> {
let (private_key, passphrase, hostname, port, host_key) = self.store.with_conn(|conn| {
let private_key = read_sealed_secret(
conn,
workspace_id,
&binding.credential_id,
credential.current_revision,
credential_id,
credential_revision,
"private_key",
)?
.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Repository SSH credential `{}` is missing its private-key revision",
binding.credential_id
"Repository SSH credential `{credential_id}` revision {credential_revision} is unavailable"
))
})?;
let passphrase = read_sealed_secret(
conn,
workspace_id,
&binding.credential_id,
credential.current_revision,
credential_id,
credential_revision,
"passphrase",
)?;
Ok((private_key, passphrase))
let (hostname, port, host_key) = conn
.query_row(
r#"SELECT h.hostname, h.port, v.host_key
FROM repository_ssh_host_trusts h
JOIN repository_ssh_host_trust_revisions v
ON v.workspace_id = h.workspace_id
AND v.host_trust_id = h.host_trust_id
WHERE h.workspace_id = ?1 AND h.host_trust_id = ?2
AND v.revision = ?3"#,
params![workspace_id, host_trust_id, host_trust_revision as i64],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)? as u16,
row.get::<_, String>(2)?,
))
},
)
.optional()?
.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Repository SSH host trust `{host_trust_id}` revision {host_trust_revision} is unavailable"
))
})?;
Ok((private_key, passphrase, hostname, port, host_key))
})?;
let private_key = self.unseal(
workspace_id,
&binding.credential_id,
credential.current_revision,
credential_id,
credential_revision,
"private_key",
private_key,
)?;
@@ -899,8 +939,8 @@ impl RepositorySecretService {
.map(|secret| {
self.unseal(
workspace_id,
&binding.credential_id,
credential.current_revision,
credential_id,
credential_revision,
"passphrase",
secret,
)
@@ -933,18 +973,18 @@ impl RepositorySecretService {
let private_key = key
.to_openssh(LineEnding::LF)
.map_err(|_| Error::Store("Repository SSH private key encoding failed".to_string()))?;
let host = if host_trust.port == 22 {
host_trust.hostname.clone()
let host = if port == 22 {
hostname
} else {
format!("[{}]:{}", host_trust.hostname, host_trust.port)
format!("[{hostname}]:{port}")
};
Ok(LeasedRepositorySshAccess {
credential_id: binding.credential_id.clone(),
credential_revision: credential.current_revision,
host_trust_id: binding.host_trust_id.clone(),
host_trust_revision: host_trust.current_revision,
credential_id: credential_id.to_string(),
credential_revision,
host_trust_id: host_trust_id.to_string(),
host_trust_revision,
private_key,
known_hosts_entry: format!("{host} {}\n", host_trust.host_key),
known_hosts_entry: format!("{host} {host_key}\n"),
})
}
@@ -1888,6 +1928,18 @@ mod tests {
.known_hosts_entry
.starts_with("example.test ssh-ed25519 ")
);
let exact = service
.lease_ssh_materialization_access_revision(
"workspace-a",
"deploy",
lease.credential_revision,
"example",
lease.host_trust_revision,
)
.unwrap();
assert_eq!(exact.credential_revision, lease.credential_revision);
assert_eq!(exact.host_trust_revision, lease.host_trust_revision);
assert_eq!(exact.known_hosts_entry, lease.known_hosts_entry);
let unknown = config_state(
r#"{
crates/workspace-server/src/server.rs
+190 -10
View File
@@ -8474,6 +8474,34 @@ async fn create_workspace_working_directory(
)
.into());
}
if let Some(existing) = api
.config_store
.load_workdir_create_operation(workspace_id, &operation_id)?
&& let (Some(kind), Some(uri), Some(revision), Some(fingerprint)) = (
existing.source_kind.as_deref(),
existing.source_uri,
existing.source_revision,
existing.source_fingerprint,
)
{
let kind = match kind {
"local_path" => workspace_api::RepositorySourceKind::LocalPath,
"file" => workspace_api::RepositorySourceKind::File,
"https" => workspace_api::RepositorySourceKind::Https,
"http" => workspace_api::RepositorySourceKind::Http,
"ssh" => workspace_api::RepositorySourceKind::Ssh,
"invalid" => workspace_api::RepositorySourceKind::Invalid,
_ => {
return Err(settings_bad_request(
"working_directory_repository_source_invalid",
"persisted Workdir create Repository source kind is invalid",
));
}
};
working_directory_request.repository.source = workspace_api::RepositorySource { kind, uri };
working_directory_request.repository.source_revision = revision;
working_directory_request.repository.source_fingerprint = fingerprint;
}
let selector = working_directory_request
.repository
.selector
@@ -8538,6 +8566,28 @@ async fn create_workspace_working_directory(
resolved_runtime_id,
config_revision: runtime_projection.config_revision,
config_projection_digest: runtime_projection.projection_digest,
source_kind: Some(
working_directory_request
.repository
.source
.kind
.as_str()
.to_string(),
),
source_uri: Some(working_directory_request.repository.source.uri.clone()),
source_revision: Some(working_directory_request.repository.source_revision),
source_fingerprint: Some(
working_directory_request
.repository
.source_fingerprint
.clone(),
),
credential_id: None,
credential_revision: None,
host_trust_id: None,
host_trust_revision: None,
repository_access_mode: None,
cache_generation: 0,
working_directory_id: next_backend_workdir_id(&request.repository_id),
state: "pending".to_string(),
failure: None,
@@ -8628,12 +8678,10 @@ async fn create_workspace_working_directory(
));
}
let repository_access_projection = active_repository_access_projection(api, workspace_id)?;
if let Err(error) = authorize_repository_materialization(
if let Err(error) = authorize_repository_materialization_operation(
api,
&reserved.resolved_runtime_id,
&operation_id,
&repository_access_projection,
&reserved,
&request_fingerprint,
&mut working_directory_request,
) {
api.config_store.finish_workdir_create_operation(
@@ -13968,6 +14016,132 @@ fn validate_working_directory_claim_for_browser(
Ok(())
}
fn authorize_repository_materialization_operation(
api: &WorkspaceApi,
operation: &WorkdirCreateOperationRecord,
request_fingerprint: &str,
request: &mut WorkingDirectoryRequest,
) -> ApiResult<()> {
let context = if request.repository.source.kind == workspace_api::RepositorySourceKind::Ssh {
if let (
Some(credential_id),
Some(credential_revision),
Some(host_trust_id),
Some(host_trust_revision),
Some(access_mode),
) = (
operation.credential_id.as_deref(),
operation.credential_revision,
operation.host_trust_id.as_deref(),
operation.host_trust_revision,
operation.repository_access_mode.as_deref(),
) {
let lease = api
.repository_secrets
.lease_ssh_materialization_access_revision(
&api.config.workspace_id,
credential_id,
credential_revision,
host_trust_id,
host_trust_revision,
)?;
let access = match access_mode {
"read_only" => workspace_api::RepositoryAccessMode::ReadOnly,
"read_write" => workspace_api::RepositoryAccessMode::ReadWrite,
_ => {
return Err(settings_bad_request(
"working_directory_repository_access_invalid",
"persisted Repository access mode is invalid",
));
}
};
RepositoryMaterializationContext {
workspace_id: api.config.workspace_id.clone(),
runtime_id: operation.resolved_runtime_id.clone(),
operation_id: operation.operation_id.clone(),
config_revision: operation.config_revision,
config_projection_digest: operation.config_projection_digest.clone(),
cache_generation: operation.cache_generation,
ssh: Some(RepositorySshMaterializationAccess {
credential_id: lease.credential_id,
credential_revision: lease.credential_revision,
host_trust_id: lease.host_trust_id,
host_trust_revision: lease.host_trust_revision,
access,
expires_at_epoch_seconds: repository_access_expiry(),
private_key: SensitiveString::new(lease.private_key.as_str()),
known_hosts_entry: SensitiveString::new(lease.known_hosts_entry),
}),
}
} else {
let projection = active_repository_access_projection(api, &api.config.workspace_id)?;
if projection.config_revision != operation.config_revision
|| projection.projection_digest != operation.config_projection_digest
{
return Err(settings_bad_request(
"working_directory_repository_access_revision_changed",
"Workspace Repository access revision changed before operation reservation was bound",
));
}
authorize_repository_materialization(
api,
&operation.resolved_runtime_id,
&operation.operation_id,
&projection,
request,
)?;
let context = request.materialization.take().ok_or_else(|| {
settings_bad_request(
"working_directory_remote_repository_access_required",
"SSH Repository access authority is unavailable",
)
})?;
let ssh = context.ssh.as_ref().ok_or_else(|| {
settings_bad_request(
"working_directory_remote_repository_access_required",
"SSH Repository access authority is unavailable",
)
})?;
api.config_store.bind_workdir_create_repository_access(
&api.config.workspace_id,
&operation.operation_id,
request_fingerprint,
&ssh.credential_id,
ssh.credential_revision,
&ssh.host_trust_id,
ssh.host_trust_revision,
match ssh.access {
workspace_api::RepositoryAccessMode::ReadOnly => "read_only",
workspace_api::RepositoryAccessMode::ReadWrite => "read_write",
},
context.cache_generation,
&now_registry_timestamp(),
)?;
context
}
} else {
RepositoryMaterializationContext {
workspace_id: api.config.workspace_id.clone(),
runtime_id: operation.resolved_runtime_id.clone(),
operation_id: operation.operation_id.clone(),
config_revision: operation.config_revision,
config_projection_digest: operation.config_projection_digest.clone(),
cache_generation: operation.cache_generation,
ssh: None,
}
};
request.materialization = Some(context);
Ok(())
}
fn repository_access_expiry() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
.saturating_add(300)
}
fn authorize_repository_materialization(
api: &WorkspaceApi,
runtime_id: &str,
@@ -13995,11 +14169,7 @@ fn authorize_repository_materialization(
host_trust_id: lease.host_trust_id,
host_trust_revision: lease.host_trust_revision,
access: binding.access,
expires_at_epoch_seconds: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
.saturating_add(300),
expires_at_epoch_seconds: repository_access_expiry(),
private_key: SensitiveString::new(lease.private_key.as_str()),
known_hosts_entry: SensitiveString::new(lease.known_hosts_entry),
})
@@ -21247,6 +21417,16 @@ mod tests {
resolved_runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
config_revision: 1,
config_projection_digest: "sha256:test".to_string(),
source_kind: Some("local_path".to_string()),
source_uri: Some("/tmp/repo".to_string()),
source_revision: Some(1),
source_fingerprint: Some("sha256:test".to_string()),
credential_id: None,
credential_revision: None,
host_trust_id: None,
host_trust_revision: None,
repository_access_mode: None,
cache_generation: 0,
working_directory_id: "workdir-provider-rejection".to_string(),
state: "pending".to_string(),
failure: None,
crates/workspace-server/src/store.rs
+110 -23
View File
@@ -257,6 +257,11 @@ const MIGRATIONS: &[Migration] = &[
name: "create Workspace Repository SSH secret authority",
apply: create_repository_ssh_secret_authority,
},
Migration {
version: 47,
name: "bind Workdir create repository access evidence",
apply: bind_workdir_create_repository_access_evidence,
},
];
struct Migration {
@@ -590,6 +595,16 @@ pub struct WorkdirCreateOperationRecord {
pub resolved_runtime_id: String,
pub config_revision: u64,
pub config_projection_digest: String,
pub source_kind: Option<String>,
pub source_uri: Option<String>,
pub source_revision: Option<u64>,
pub source_fingerprint: Option<String>,
pub credential_id: Option<String>,
pub credential_revision: Option<u64>,
pub host_trust_id: Option<String>,
pub host_trust_revision: Option<u64>,
pub repository_access_mode: Option<String>,
pub cache_generation: u64,
pub working_directory_id: String,
pub state: String,
pub failure: Option<String>,
@@ -6812,6 +6827,25 @@ fn create_repository_ssh_secret_authority(conn: &Connection) -> Result<()> {
Ok(())
}
fn bind_workdir_create_repository_access_evidence(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
ALTER TABLE workdir_create_operations ADD COLUMN source_kind TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN source_uri TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN source_revision INTEGER;
ALTER TABLE workdir_create_operations ADD COLUMN source_fingerprint TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN credential_id TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN credential_revision INTEGER;
ALTER TABLE workdir_create_operations ADD COLUMN host_trust_id TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN host_trust_revision INTEGER;
ALTER TABLE workdir_create_operations ADD COLUMN repository_access_mode TEXT;
ALTER TABLE workdir_create_operations
ADD COLUMN cache_generation INTEGER NOT NULL DEFAULT 0;
"#,
)?;
Ok(())
}
fn create_workspace_catalog_operations(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
@@ -9697,7 +9731,7 @@ mod tests {
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
let remote = conn
.query_row(
"SELECT source_kind, source_uri, source_revision, source_fingerprint, observed_status \
@@ -9775,7 +9809,7 @@ mod tests {
let before = std::fs::read(&path).unwrap();
let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
assert_eq!(plan.current_schema_version, 36);
assert_eq!(plan.target_schema_version, 46);
assert_eq!(plan.target_schema_version, 47);
assert!(plan.migration_required);
assert_eq!(plan.worker_count, 1);
assert_eq!(plan.mappings[0].legacy_worker_id, 7);
@@ -9789,7 +9823,7 @@ mod tests {
store
.with_conn(|conn| {
assert!(table_exists(conn, "worker_diagnostics_archives")?);
assert_eq!(current_schema_version(conn)?, 46);
assert_eq!(current_schema_version(conn)?, 47);
Ok(())
})
.unwrap();
@@ -9925,7 +9959,7 @@ mod tests {
),
]
);
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
let foreign_key_error: Option<String> = conn
.query_row("PRAGMA foreign_key_check", [], |row| row.get(0))
.optional()
@@ -10054,7 +10088,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap());
let controller_worker_id: String = conn
.query_row(
@@ -10172,7 +10206,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
}
@@ -10190,7 +10224,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
let settings = conn
.query_row(
"SELECT settings_revision, language FROM workspace_memory_settings \
@@ -10231,7 +10265,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(table_exists(&conn, "flow_sources").unwrap());
assert!(table_exists(&conn, "flow_source_revisions").unwrap());
assert!(!table_exists(&conn, "flow_instances").unwrap());
@@ -10298,7 +10332,7 @@ INSERT INTO worker_workdir_attachment_reservations (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
let repositories_sql: String = conn
.query_row(
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
@@ -10481,7 +10515,7 @@ INSERT INTO workdir_registry (
let db = dir.path().join("control-plane.sqlite");
let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46);
assert_eq!(store.schema_version().await.unwrap(), 47);
assert!(
!store
.with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -10498,7 +10532,7 @@ INSERT INTO workdir_registry (
store.upsert_workspace(&record).await.unwrap();
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(reopened.schema_version().await.unwrap(), 46);
assert_eq!(reopened.schema_version().await.unwrap(), 47);
assert_eq!(
reopened.get_workspace("local-dev").await.unwrap(),
Some(record)
@@ -11252,7 +11286,7 @@ INSERT INTO worker_registry (
let migrated = SqliteWorkspaceStore::open(&db_path).unwrap();
migrated
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 46);
assert_eq!(current_schema_version(conn)?, 47);
assert_eq!(
conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?,
1,
@@ -11609,13 +11643,13 @@ INSERT INTO worker_registry (
DROP TABLE repository_ssh_host_trust_revisions;
DROP TABLE repository_ssh_host_trusts;
DROP TABLE workdir_create_operations;
DELETE FROM __yoi_schema_migrations WHERE version IN (45, 46);",
DELETE FROM __yoi_schema_migrations WHERE version IN (45, 46, 47);",
)
.unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 44);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(table_exists(&conn, "workdir_create_operations").unwrap());
let columns = table_columns(&conn, "workdir_create_operations").unwrap();
for required in [
@@ -11647,13 +11681,23 @@ INSERT INTO worker_registry (
DROP TABLE repository_ssh_credentials;
DROP TABLE repository_ssh_host_trust_revisions;
DROP TABLE repository_ssh_host_trusts;
DELETE FROM __yoi_schema_migrations WHERE version = 46;",
ALTER TABLE workdir_create_operations DROP COLUMN source_kind;
ALTER TABLE workdir_create_operations DROP COLUMN source_uri;
ALTER TABLE workdir_create_operations DROP COLUMN source_revision;
ALTER TABLE workdir_create_operations DROP COLUMN source_fingerprint;
ALTER TABLE workdir_create_operations DROP COLUMN credential_id;
ALTER TABLE workdir_create_operations DROP COLUMN credential_revision;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_id;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_revision;
ALTER TABLE workdir_create_operations DROP COLUMN repository_access_mode;
ALTER TABLE workdir_create_operations DROP COLUMN cache_generation;
DELETE FROM __yoi_schema_migrations WHERE version IN (46, 47);",
)
.unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 45);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
for table in [
"repository_ssh_credentials",
"repository_ssh_credential_revisions",
@@ -11672,19 +11716,62 @@ INSERT INTO worker_registry (
assert!(foreign_key_error.is_none());
}
#[test]
fn schema_v47_binds_workdir_create_repository_access_evidence() {
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations(&conn).unwrap();
conn.execute_batch(
"ALTER TABLE workdir_create_operations DROP COLUMN source_kind;
ALTER TABLE workdir_create_operations DROP COLUMN source_uri;
ALTER TABLE workdir_create_operations DROP COLUMN source_revision;
ALTER TABLE workdir_create_operations DROP COLUMN source_fingerprint;
ALTER TABLE workdir_create_operations DROP COLUMN credential_id;
ALTER TABLE workdir_create_operations DROP COLUMN credential_revision;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_id;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_revision;
ALTER TABLE workdir_create_operations DROP COLUMN repository_access_mode;
ALTER TABLE workdir_create_operations DROP COLUMN cache_generation;
DELETE FROM __yoi_schema_migrations WHERE version = 47;",
)
.unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 47);
let columns = table_columns(&conn, "workdir_create_operations").unwrap();
for required in [
"source_kind",
"source_uri",
"source_revision",
"source_fingerprint",
"credential_id",
"credential_revision",
"host_trust_id",
"host_trust_revision",
"repository_access_mode",
"cache_generation",
] {
assert!(
columns.iter().any(|column| column == required),
"missing column {required}"
);
}
}
#[test]
fn server_refuses_a_database_from_a_newer_schema_generation() {
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations(&conn).unwrap();
conn.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (47, 'future')",
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (48, 'future')",
[],
)
.unwrap();
let error = apply_migrations(&conn).unwrap_err().to_string();
assert!(error.contains("schema version 47 is newer"), "{error}");
assert!(error.contains("schema version 48 is newer"), "{error}");
assert!(error.contains("refusing to serve"), "{error}");
}
@@ -11905,7 +11992,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
apply_migrations(&mut conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
assert_eq!(current_schema_version(&conn).unwrap(), 47);
let workspace_id: Option<String> = conn
.query_row(
"SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'",
@@ -12528,7 +12615,7 @@ WHERE workspace_id = 'workspace-a'
.unwrap();
let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46);
assert_eq!(store.schema_version().await.unwrap(), 47);
store
.with_conn(|conn| {
@@ -12717,7 +12804,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn repository_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46);
assert_eq!(store.schema_version().await.unwrap(), 47);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -12795,7 +12882,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn memory_authority_records_round_trip_and_close_staging() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46);
assert_eq!(store.schema_version().await.unwrap(), 47);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -13202,7 +13289,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn account_and_login_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46);
assert_eq!(store.schema_version().await.unwrap(), 47);
let now = "2026-07-22T00:00:00Z".to_string();
let account = AccountRecord {
account_id: "acct-user-alice".to_string(),
crates/workspace-server/src/workdir_create_operations.rs
+153 -15
View File
@@ -48,9 +48,10 @@ impl SqliteWorkspaceStore {
r#"INSERT OR IGNORE INTO workdir_create_operations (
workspace_id, operation_id, request_fingerprint, repository_id, selector,
requested_runtime_id, resolved_runtime_id, config_revision,
config_projection_digest, working_directory_id, state, failure,
config_projection_digest, source_kind, source_uri, source_revision,
source_fingerprint, working_directory_id, state, failure,
created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)"#,
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)"#,
params![
record.workspace_id,
record.operation_id,
@@ -61,6 +62,10 @@ impl SqliteWorkspaceStore {
record.resolved_runtime_id,
record.config_revision as i64,
record.config_projection_digest,
record.source_kind,
record.source_uri,
record.source_revision.map(|revision| revision as i64),
record.source_fingerprint,
record.working_directory_id,
record.state,
record.failure,
@@ -87,6 +92,81 @@ impl SqliteWorkspaceStore {
})
}
pub fn bind_workdir_create_repository_access(
&self,
workspace_id: &str,
operation_id: &str,
request_fingerprint: &str,
credential_id: &str,
credential_revision: u64,
host_trust_id: &str,
host_trust_revision: u64,
repository_access_mode: &str,
cache_generation: u64,
now: &str,
) -> Result<WorkdirCreateOperationRecord> {
self.with_conn_mut(|conn| {
let operation = read_workdir_create_operation(conn, workspace_id, operation_id)?
.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Workdir create operation `{operation_id}` disappeared before Repository access binding"
))
})?;
if operation.request_fingerprint != request_fingerprint {
return Err(Error::InvalidInput(format!(
"Workdir create operation `{operation_id}` was reused with different input"
)));
}
if let Some(existing) = operation.credential_id.as_deref() {
if existing != credential_id
|| operation.credential_revision != Some(credential_revision)
|| operation.host_trust_id.as_deref() != Some(host_trust_id)
|| operation.host_trust_revision != Some(host_trust_revision)
|| operation.repository_access_mode.as_deref()
!= Some(repository_access_mode)
|| operation.cache_generation != cache_generation
{
return Err(Error::InvalidInput(format!(
"Workdir create operation `{operation_id}` Repository access evidence changed"
)));
}
return Ok(operation);
}
conn.execute(
r#"UPDATE workdir_create_operations
SET credential_id = ?4, credential_revision = ?5,
host_trust_id = ?6, host_trust_revision = ?7,
repository_access_mode = ?8, cache_generation = ?9,
updated_at = ?10
WHERE workspace_id = ?1 AND operation_id = ?2
AND request_fingerprint = ?3 AND credential_id IS NULL"#,
params![
workspace_id,
operation_id,
request_fingerprint,
credential_id,
i64::try_from(credential_revision).map_err(|_| Error::InvalidInput(
"credential revision is out of range".to_string()
))?,
host_trust_id,
i64::try_from(host_trust_revision).map_err(|_| Error::InvalidInput(
"host-trust revision is out of range".to_string()
))?,
repository_access_mode,
i64::try_from(cache_generation).map_err(|_| Error::InvalidInput(
"cache generation is out of range".to_string()
))?,
now,
],
)?;
read_workdir_create_operation(conn, workspace_id, operation_id)?.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Workdir create operation `{operation_id}` disappeared after Repository access binding"
))
})
})
}
pub fn finish_workdir_create_operation(
&self,
workspace_id: &str,
@@ -141,7 +221,10 @@ fn read_workdir_create_operation(
conn.query_row(
r#"SELECT workspace_id, operation_id, request_fingerprint, repository_id, selector,
requested_runtime_id, resolved_runtime_id, config_revision,
config_projection_digest, working_directory_id, state, failure,
config_projection_digest, source_kind, source_uri, source_revision,
source_fingerprint, credential_id, credential_revision,
host_trust_id, host_trust_revision, repository_access_mode,
cache_generation, working_directory_id, state, failure,
created_at, updated_at
FROM workdir_create_operations
WHERE workspace_id = ?1 AND operation_id = ?2"#,
@@ -157,11 +240,21 @@ fn read_workdir_create_operation(
resolved_runtime_id: row.get(6)?,
config_revision: row.get::<_, i64>(7)? as u64,
config_projection_digest: row.get(8)?,
working_directory_id: row.get(9)?,
state: row.get(10)?,
failure: row.get(11)?,
created_at: row.get(12)?,
updated_at: row.get(13)?,
source_kind: row.get(9)?,
source_uri: row.get(10)?,
source_revision: row.get::<_, Option<i64>>(11)?.map(|value| value as u64),
source_fingerprint: row.get(12)?,
credential_id: row.get(13)?,
credential_revision: row.get::<_, Option<i64>>(14)?.map(|value| value as u64),
host_trust_id: row.get(15)?,
host_trust_revision: row.get::<_, Option<i64>>(16)?.map(|value| value as u64),
repository_access_mode: row.get(17)?,
cache_generation: row.get::<_, i64>(18)? as u64,
working_directory_id: row.get(19)?,
state: row.get(20)?,
failure: row.get(21)?,
created_at: row.get(22)?,
updated_at: row.get(23)?,
})
},
)
@@ -222,6 +315,16 @@ mod tests {
resolved_runtime_id: "arcadia".to_string(),
config_revision: 7,
config_projection_digest: "sha256:projection".to_string(),
source_kind: Some("local_path".to_string()),
source_uri: Some("/tmp/repo".to_string()),
source_revision: Some(1),
source_fingerprint: Some("sha256:source".to_string()),
credential_id: None,
credential_revision: None,
host_trust_id: None,
host_trust_revision: None,
repository_access_mode: None,
cache_generation: 0,
working_directory_id: "wd-1".to_string(),
state: "pending".to_string(),
failure: None,
@@ -232,20 +335,55 @@ mod tests {
store.reserve_workdir_create_operation(&record).unwrap(),
record
);
let bound = store
.bind_workdir_create_repository_access(
"workspace",
"call-1",
&record.request_fingerprint,
"credential-1",
3,
"trust-1",
5,
"read_only",
2,
"2026-08-24T00:00:01Z",
)
.unwrap();
assert_eq!(bound.credential_id.as_deref(), Some("credential-1"));
assert_eq!(bound.credential_revision, Some(3));
assert_eq!(bound.host_trust_revision, Some(5));
assert_eq!(bound.cache_generation, 2);
assert!(
store
.bind_workdir_create_repository_access(
"workspace",
"call-1",
&record.request_fingerprint,
"credential-1",
4,
"trust-1",
5,
"read_only",
2,
"2026-08-24T00:00:02Z",
)
.is_err()
);
let mut changed_resolution = record.clone();
changed_resolution.resolved_runtime_id = "other".to_string();
changed_resolution.config_revision = 8;
assert_eq!(
store
.reserve_workdir_create_operation(&changed_resolution)
.unwrap(),
record
);
changed_resolution.source_uri = Some("ssh://git@other.test/repo.git".to_string());
changed_resolution.source_revision = Some(9);
let replayed = store
.reserve_workdir_create_operation(&changed_resolution)
.unwrap();
assert_eq!(replayed, bound);
assert_eq!(replayed.source_uri.as_deref(), Some("/tmp/repo"));
assert_eq!(
store
.load_workdir_create_operation("workspace", "call-1")
.unwrap(),
Some(record.clone())
Some(bound.clone())
);
let mut changed_input = record.clone();
changed_input.request_fingerprint =