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

feat: observe repository refs through runtime providers

Browse Source
This commit is contained in:
Hare
2026-09-03 12:56:06 +09:00
parent 6b1b8a8846
commit 56798f9fb4
6 changed files with 366 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+24
View File
@@ -179,6 +179,30 @@ pub struct WorkingDirectoryRequest {
pub materialization: Option<RepositoryMaterializationContext>,
}
/// Backend-authorized request to freshly resolve one Repository provider ref.
///
/// Runtime executes this against the registered source itself rather than a Workdir
/// or Runtime cache. Secret material is fetched through `materialization` and never
/// appears in the result.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepositoryRefObservationRequest {
pub repository: WorkingDirectoryRepository,
pub selector: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub materialization: Option<RepositoryMaterializationContext>,
}
/// Provider-neutral proof of one freshly observed Repository ref.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepositoryRefObservation {
pub repository_id: String,
pub source_revision: u64,
pub source_fingerprint: String,
pub selector: String,
pub revision_ref: String,
pub observed_at_epoch_seconds: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryClaim {
pub working_directory_id: String,
crates/worker-runtime/src/execution.rs
+18
View File
@@ -1,4 +1,5 @@
use crate::catalog::{
RepositoryRefObservation, RepositoryRefObservationRequest,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
};
use crate::config_bundle::ConfigBundle;
@@ -333,6 +334,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
))
}
fn observe_repository_ref(
&self,
_request: &RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
Err(WorkingDirectoryDiagnostic::rejected(
"repository_ref_provider_unavailable",
"Worker execution backend does not support Repository ref observation",
))
}
fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
Vec::new()
}
@@ -501,6 +512,13 @@ impl WorkerExecutionBackendRef {
.authorize_working_directory_repository_access(request)
}
pub(crate) fn observe_repository_ref(
&self,
request: &RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
self.backend.observe_repository_ref(request)
}
pub(crate) fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
self.backend.list_working_directories()
}
crates/worker-runtime/src/http_server.rs
+37 -4
View File
@@ -11,9 +11,9 @@ use crate::auth::{
verify_capability_token,
};
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkspaceApiRef,
ConfigBundleRef, CreateWorkerRequest, RepositoryRefObservationRequest, WorkerDetail,
WorkerLifecycleAck, WorkerSummary, WorkingDirectoryRepositoryAccessRequest,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkspaceApiRef,
};
use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
use crate::error::RuntimeError;
@@ -208,6 +208,7 @@ fn runtime_http_router_with_optional_auth(
"/v1/working-directories/repository-access",
post(authorize_working_directory_repository_access),
)
.route("/v1/repository-refs/observe", post(observe_repository_ref))
.route(
"/v1/working-directories/{working_directory_id}/sessions",
post(open_workdir_session),
@@ -583,6 +584,31 @@ async fn authorize_working_directory_repository_access(
}))
}
async fn observe_repository_ref(
State(state): State<RuntimeHttpState>,
Extension(auth): Extension<RuntimeAuthContext>,
body: Result<Json<RepositoryRefObservationRequest>, JsonRejection>,
) -> RestResult<crate::catalog::RepositoryRefObservation> {
let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
if request
.materialization
.as_ref()
.is_some_and(|materialization| materialization.workspace_id != auth.workspace_id)
{
return Err(RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"repository_ref_observation_workspace_mismatch",
"Repository ref observation authority does not match the authenticated Workspace",
));
}
let observation = state
.runtime
.observe_repository_ref_from_resource(request)
.await
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(observation))
}
async fn list_working_directories(
State(state): State<RuntimeHttpState>,
) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
@@ -1750,7 +1776,10 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
if path == "/v1/workers" && *method == Method::POST {
return Some("workers:create");
}
if path == "/v1/working-directories/repository-access" && *method == Method::POST {
if (path == "/v1/working-directories/repository-access"
|| path == "/v1/repository-refs/observe")
&& *method == Method::POST
{
return Some("workdirs:operate");
}
if path.starts_with("/v1/workdir-sessions")
@@ -2424,6 +2453,10 @@ mod tests {
required_runtime_permission(&Method::POST, "/v1/working-directories/repository-access",),
Some("workdirs:operate")
);
assert_eq!(
required_runtime_permission(&Method::POST, "/v1/repository-refs/observe"),
Some("workdirs:operate")
);
assert_eq!(
required_runtime_permission(&Method::POST, "/v1/working-directories/wd-1/sessions"),
Some("workdirs:operate")
crates/worker-runtime/src/runtime.rs
+28 -2
View File
@@ -1,6 +1,7 @@
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck,
WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, RepositoryRefObservation,
RepositoryRefObservationRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus, WorkerSummary,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef,
};
use crate::config_bundle::{
@@ -392,6 +393,31 @@ impl Runtime {
self.create_working_directory(request)
}
pub async fn observe_repository_ref_from_resource(
&self,
mut request: RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, RuntimeError> {
if let Some(ssh) = request
.materialization
.as_mut()
.and_then(|materialization| materialization.ssh.as_mut())
{
self.resolve_repository_access_resource(ssh).await?;
}
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "Repository ref observation requires an execution backend".to_string(),
}
})?
};
backend
.observe_repository_ref(&request)
.map_err(RuntimeError::from)
}
pub fn authorize_working_directory_repository_access(
&self,
request: WorkingDirectoryRepositoryAccessRequest,
crates/worker-runtime/src/worker_backend.rs
+14
View File
@@ -20,6 +20,7 @@ use crate::auth::{
};
use crate::catalog::{
CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource,
RepositoryRefObservation, RepositoryRefObservationRequest,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
};
use crate::execution::{
@@ -1645,6 +1646,19 @@ where
materializer.authorize_repository_access(request)
}
fn observe_repository_ref(
&self,
request: &RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
let materializer = self.working_directory_materializer.as_ref().ok_or_else(|| {
WorkingDirectoryDiagnostic::rejected(
"repository_ref_provider_unavailable",
"Repository ref observation requested, but no materializer is configured for this Runtime backend",
)
})?;
materializer.observe_repository_ref(request)
}
fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
self.working_directory_materializer
.as_ref()
crates/worker-runtime/src/working_directory.rs
+245 -1
View File
@@ -1,5 +1,6 @@
use crate::catalog::{
MaterializerKind, RepositorySshMaterializationAccess, WorkingDirectoryCleanupTarget,
MaterializerKind, RepositoryRefObservation, RepositoryRefObservationRequest,
RepositorySshMaterializationAccess, WorkingDirectoryCleanupTarget,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectoryStatusKind, WorkingDirectorySummary,
};
@@ -196,6 +197,11 @@ pub trait WorkingDirectoryMaterializer: Send + Sync + 'static {
request: &WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), WorkingDirectoryDiagnostic>;
fn observe_repository_ref(
&self,
request: &RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic>;
fn bind_working_directory(
&self,
working_directory_id: &str,
@@ -943,6 +949,78 @@ impl WorkingDirectoryMaterializer for RuntimeGitCacheMaterializer {
self.cache_repository_access(&request.working_directory_id, ssh)
}
fn observe_repository_ref(
&self,
request: &RepositoryRefObservationRequest,
) -> Result<RepositoryRefObservation, WorkingDirectoryDiagnostic> {
let selector = request.selector.trim();
validate_selector(selector)?;
if !selector.starts_with("refs/heads/") {
return Err(WorkingDirectoryDiagnostic::new(
"repository_ref_selector_invalid",
"Repository ref observation requires an exact branch selector",
));
}
let working_request = WorkingDirectoryRequest {
repository: request.repository.clone(),
materializer: MaterializerKind::RuntimeGitCache,
backend_workdir_id: None,
materialization: request.materialization.clone(),
};
Self::validate_request(&working_request)?;
let access = RepositoryCommandAccess::prepare(&self.runtime_root, &working_request)?;
let mut command = repository_git_command(&working_request, access.as_ref());
command.args([
"ls-remote",
"--exit-code",
"--refs",
request.repository.source.uri.as_str(),
selector,
]);
let output = run_repository_git_stdout(command, request.repository.source.kind)?;
let mut lines = output.lines();
let line = lines.next().ok_or_else(|| {
WorkingDirectoryDiagnostic::new(
"repository_ref_not_found",
"Repository provider did not return the requested ref",
)
})?;
if lines.next().is_some() {
return Err(WorkingDirectoryDiagnostic::new(
"repository_ref_response_invalid",
"Repository provider returned an ambiguous ref observation",
));
}
let (revision_ref, observed_selector) = line.split_once('\t').ok_or_else(|| {
WorkingDirectoryDiagnostic::new(
"repository_ref_response_invalid",
"Repository provider returned an invalid ref observation",
)
})?;
if observed_selector != selector
|| !matches!(revision_ref.len(), 40 | 64)
|| !revision_ref.bytes().all(|byte| byte.is_ascii_hexdigit())
{
return Err(WorkingDirectoryDiagnostic::new(
"repository_ref_response_invalid",
"Repository provider returned an invalid ref observation",
));
}
Ok(RepositoryRefObservation {
repository_id: request.repository.id.clone(),
source_revision: request.repository.source_revision,
source_fingerprint: request.repository.source_fingerprint.clone(),
selector: selector.to_string(),
revision_ref: revision_ref.to_ascii_lowercase(),
observed_at_epoch_seconds: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs(),
})
}
fn bind_working_directory(
&self,
working_directory_id: &str,
@@ -2022,6 +2100,112 @@ fn repository_git_command(
command
}
fn read_bounded_command_output(mut reader: impl Read) -> Vec<u8> {
const MAX_CAPTURE_BYTES: usize = 8192;
let mut captured = Vec::new();
let mut chunk = [0_u8; 4096];
loop {
match reader.read(&mut chunk) {
Ok(0) | Err(_) => break,
Ok(read) => {
let remaining = MAX_CAPTURE_BYTES.saturating_sub(captured.len());
captured.extend_from_slice(&chunk[..read.min(remaining)]);
}
}
}
captured
}
fn run_repository_git_stdout(
mut command: Command,
source_kind: workspace_api::RepositorySourceKind,
) -> Result<String, WorkingDirectoryDiagnostic> {
command
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let mut child = command.spawn().map_err(|_| {
WorkingDirectoryDiagnostic::new(
"repository_ref_provider_unavailable",
"Repository provider operation could not be started",
)
})?;
let stdout = child.stdout.take().ok_or_else(|| {
WorkingDirectoryDiagnostic::new(
"repository_ref_provider_unavailable",
"Repository provider response could not be captured",
)
})?;
let stderr = child.stderr.take().ok_or_else(|| {
WorkingDirectoryDiagnostic::new(
"repository_ref_provider_unavailable",
"Repository provider diagnostic could not be captured",
)
})?;
let stdout_reader = std::thread::spawn(move || read_bounded_command_output(stdout));
let stderr_reader = std::thread::spawn(move || read_bounded_command_output(stderr));
let started = Instant::now();
let status = loop {
if let Some(status) = child.try_wait().map_err(|_| {
WorkingDirectoryDiagnostic::new(
"repository_ref_provider_unavailable",
"Repository provider operation status could not be observed",
)
})? {
break status;
}
if started.elapsed() >= REPOSITORY_COMMAND_TIMEOUT {
let _ = child.kill();
let _ = child.wait();
let _ = stdout_reader.join();
let _ = stderr_reader.join();
return Err(WorkingDirectoryDiagnostic::new(
"repository_ref_provider_timeout",
"Repository provider operation exceeded the Runtime time limit",
));
}
std::thread::sleep(Duration::from_millis(25));
};
let stdout = stdout_reader.join().unwrap_or_default();
let stderr = stderr_reader.join().unwrap_or_default();
if status.success() {
return String::from_utf8(stdout).map_err(|_| {
WorkingDirectoryDiagnostic::new(
"repository_ref_response_invalid",
"Repository provider returned a non-UTF-8 ref observation",
)
});
}
if status.code() == Some(2) {
return Err(WorkingDirectoryDiagnostic::new(
"repository_ref_not_found",
"Repository provider did not return the requested ref",
));
}
let diagnostic = String::from_utf8_lossy(&stderr).to_ascii_lowercase();
let auth_failed = source_kind.is_remote()
&& [
"authentication failed",
"permission denied",
"could not read username",
"publickey",
]
.iter()
.any(|marker| diagnostic.contains(marker));
Err(WorkingDirectoryDiagnostic::new(
if auth_failed {
"repository_ref_provider_auth_failed"
} else {
"repository_ref_provider_unavailable"
},
if auth_failed {
"Repository provider rejected the operation-scoped authentication"
} else {
"Repository provider operation failed"
},
))
}
fn run_repository_git(
mut command: Command,
code: &'static str,
@@ -2493,6 +2677,66 @@ mod tests {
WorkerRef::new(WorkerId::from_legacy_u64(sequence))
}
#[test]
fn repository_ref_observation_reads_the_provider_fresh() {
let repo = create_clean_repo();
git(repo.path(), &["branch", "published"]);
let runtime_root = tempfile::tempdir().unwrap();
let materializer = RuntimeGitCacheMaterializer::new(runtime_root.path());
let repository = request(repo.path()).repository;
let observation_request = RepositoryRefObservationRequest {
repository,
selector: "refs/heads/published".to_string(),
materialization: None,
};
let first = materializer
.observe_repository_ref(&observation_request)
.unwrap();
assert_eq!(
first.revision_ref,
git_stdout(repo.path(), ["rev-parse", "published"]).unwrap()
);
fs::write(repo.path().join("second.txt"), "second\n").unwrap();
git(repo.path(), &["add", "second.txt"]);
git(repo.path(), &["commit", "-m", "second"]);
git(repo.path(), &["branch", "-f", "published"]);
let second = materializer
.observe_repository_ref(&observation_request)
.unwrap();
assert_ne!(first.revision_ref, second.revision_ref);
assert_eq!(
second.revision_ref,
git_stdout(repo.path(), ["rev-parse", "published"]).unwrap()
);
}
#[test]
fn repository_ref_observation_rejects_missing_and_non_branch_selectors() {
let repo = create_clean_repo();
let runtime_root = tempfile::tempdir().unwrap();
let materializer = RuntimeGitCacheMaterializer::new(runtime_root.path());
let repository = request(repo.path()).repository;
let missing = materializer
.observe_repository_ref(&RepositoryRefObservationRequest {
repository: repository.clone(),
selector: "refs/heads/not-published".to_string(),
materialization: None,
})
.unwrap_err();
assert_eq!(missing.code, "repository_ref_not_found");
let non_branch = materializer
.observe_repository_ref(&RepositoryRefObservationRequest {
repository,
selector: "HEAD".to_string(),
materialization: None,
})
.unwrap_err();
assert_eq!(non_branch.code, "repository_ref_selector_invalid");
}
#[test]
fn local_git_repo_materializes_detached_worktree_under_runtime_root() {
let repo = create_clean_repo();