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

fix(server): own remaining request callers and stop continuation

Browse Source
This commit is contained in:
Hare
2026-10-12 17:36:08 +09:00
parent 4d98119440
commit e436b2ae58
9 changed files with 1507 additions and 342 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+523 -277
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/server/worker_operations.rs
+32 -40
View File
@@ -383,46 +383,38 @@ impl WorkspaceWorker {
) -> ApiResult<WorkerLifecycleResult> {
let guard = self.authorize_operation(context, "stop").await?;
let worker = self.clone();
let (result, guard) = self
.api
.host_executors
.host()?
.run(move || -> ApiResult<_> {
let result = worker
.api
.runtime
.stop_worker(&worker.identity, request)
.map_err(|e| e.into_error())?;
Ok((result, guard))
})
.await?;
if result.state != InternalWorkerOperationState::Accepted {
return Ok(result);
}
parse_runtime_worker_id_for_registry(&self.identity.worker_id)?;
let session_lock = current_worker_session_lock(&self.api, &self.identity);
let session_guard = session_lock.lock_owned().await;
close_current_worker_session_locked(&self.api, &self.identity).await?;
let worker = self.clone();
self.api
.host_executors
.host()?
.run(move || -> ApiResult<_> {
let (_control, _session) = (guard, session_guard);
if let Some(record) = worker
.api
.store
.get_worker_registry(worker.api.workspace_id(), &worker.identity)?
{
sync_linked_workdir_after_worker_stop(
&worker.api,
&worker.identity.runtime_id,
&record,
)?;
}
Ok(result)
})
.await
// Once admitted, Runtime stop, session close and linked Workdir
// publication have one owner. Never hand an accepted stop's guards
// back to its cancellable waiter or acquire this Host lane again.
run_host_workflow(self.api.host_executors.host()?, async move {
let _control = guard;
let result = worker
.api
.runtime
.stop_worker(&worker.identity, request)
.map_err(|error| error.into_error())?;
if result.state != InternalWorkerOperationState::Accepted {
return Ok(result);
}
parse_runtime_worker_id_for_registry(&worker.identity.worker_id)?;
let _session = current_worker_session_lock(&worker.api, &worker.identity)
.lock_owned()
.await;
close_current_worker_session_locked(&worker.api, &worker.identity).await?;
if let Some(record) = worker
.api
.store
.get_worker_registry(worker.api.workspace_id(), &worker.identity)?
{
sync_linked_workdir_after_worker_stop(
&worker.api,
&worker.identity.runtime_id,
&record,
)?;
}
Ok(result)
})
.await
}
pub(super) async fn cancel(
crates/workspace-server/src/server_db_execution_tests.rs
+118 -16
View File
@@ -20,10 +20,16 @@ const SUBJECTS: &str = "/api/w/db-execution/subjektiv/subjects";
const OBSERVE_LIMIT: StdDuration = StdDuration::from_secs(2);
const WATCHDOG: StdDuration = StdDuration::from_secs(5);
async fn observed<T>(future: impl Future<Output = T>) -> T {
tokio::time::timeout(OBSERVE_LIMIT, future)
.await
.expect("request did not reach the expected DB/lifecycle state")
#[track_caller]
fn observed<T>(future: impl Future<Output = T>) -> impl Future<Output = T> {
let caller = std::panic::Location::caller();
async move {
tokio::time::timeout(OBSERVE_LIMIT, future)
.await
.unwrap_or_else(|_| {
panic!("request did not reach the expected DB/lifecycle state at {caller}")
})
}
}
struct Fixture {
@@ -954,7 +960,7 @@ impl HeldRuntimeReply {
Self::with_held_path(Some(path.to_owned()))
}
fn with_held_path(path: Option<String>) -> Self {
use std::io::{Read, Write};
use std::io::Write;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let (entered, received) = oneshot::channel();
@@ -967,16 +973,9 @@ impl HeldRuntimeReply {
return;
};
stream.set_read_timeout(Some(WATCHDOG)).unwrap();
let mut buf = [0; 4096];
let Ok(n) = stream.read(&mut buf) else { return };
if n == 0 {
let Ok(Some(request)) = read_runtime_fixture_request(&mut stream) else {
return;
}
let request = String::from_utf8_lossy(&buf[..n])
.lines()
.next()
.unwrap()
.to_owned();
};
if path.as_ref().is_some_and(|path| !request.contains(path)) {
// Ready's best-effort notification lookup precedes its detail
// source observation. Deny it promptly; hold only the named stage.
@@ -1012,6 +1011,76 @@ impl Drop for HeldRuntimeReply {
}
}
// Read one complete request before selecting a held path or denying a prior
// notification. TCP reads are not HTTP frame boundaries; replying while a POST
// body remains unread can also reset the connection instead of delivering 503.
fn read_runtime_fixture_request(reader: impl std::io::Read) -> std::io::Result<Option<String>> {
use std::io::{BufRead, Read};
let mut reader = std::io::BufReader::new(reader);
let mut line = String::new();
if reader.read_line(&mut line)? == 0 {
return Ok(None);
}
let request = line.trim_end().to_owned();
let mut header_bytes = line.len();
let mut body_len = 0;
loop {
line.clear();
if reader.read_line(&mut line)? == 0 {
return Err(std::io::ErrorKind::UnexpectedEof.into());
}
header_bytes += line.len();
if header_bytes > 16384 {
return Err(std::io::ErrorKind::InvalidData.into());
}
if line == "\r\n" {
break;
}
if let Some((name, value)) = line.split_once(':') {
if name.eq_ignore_ascii_case("content-length") {
body_len = value
.trim()
.parse::<usize>()
.map_err(|_| std::io::ErrorKind::InvalidData)?;
if body_len > 1024 * 1024 {
return Err(std::io::ErrorKind::InvalidData.into());
}
}
if name.eq_ignore_ascii_case("transfer-encoding") {
return Err(std::io::ErrorKind::InvalidData.into());
}
}
}
let mut body = vec![0; body_len];
reader.read_exact(&mut body)?;
Ok(Some(request))
}
#[test]
fn held_runtime_fixture_consumes_fragmented_headers_and_body_before_selecting_the_path() {
struct Fragmented(std::io::Cursor<Vec<u8>>);
impl std::io::Read for Fragmented {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
let length = buffer.len().min(2);
std::io::Read::read(&mut self.0, &mut buffer[..length])
}
}
let request = b"POST /v1/repository-refs/observe HTTP/1.1\r\nHost: localhost\r\nContent-Length: 5\r\n\r\nnotes";
let mut reader = Fragmented(std::io::Cursor::new(request.to_vec()));
assert_eq!(
read_runtime_fixture_request(&mut reader)
.unwrap()
.as_deref(),
Some("POST /v1/repository-refs/observe HTTP/1.1")
);
assert_eq!(reader.0.position(), request.len() as u64);
let truncated = &request[..request.len() - 1];
assert_eq!(
read_runtime_fixture_request(truncated).unwrap_err().kind(),
std::io::ErrorKind::UnexpectedEof
);
}
struct CountJobLifetime(Arc<std::sync::atomic::AtomicUsize>);
struct CountJobLease(Arc<std::sync::atomic::AtomicUsize>);
impl DbOperationLifetime for CountJobLifetime {
@@ -1538,6 +1607,31 @@ async fn cancelled_runtime_key_revoke_commits_audit_and_removes_broker_registrat
protocol::subscription::EventSubscriptionSelector::RuntimeWorkers,
)
.unwrap();
// Broker registration starts an independent cold signing DB job. Observe
// its actual handshake before occupying SQL for revoke: otherwise this
// fixture races that producer against HTTP auth in its 1+1 DB budget.
// Keep the handshake pending so downstream shutdown is still exercised.
let (mut upstream, _) = observed(listener.accept()).await.unwrap();
let handshake = observed(async {
use tokio::io::AsyncReadExt;
let mut bytes = Vec::new();
while !bytes.ends_with(b"\r\n\r\n") {
let mut chunk = [0; 1024];
let count = upstream.read(&mut chunk).await.unwrap();
assert_ne!(count, 0, "broker closed before sending its handshake");
bytes.extend_from_slice(&chunk[..count]);
assert!(bytes.len() <= 8192, "unexpectedly large handshake");
}
String::from_utf8(bytes).unwrap()
})
.await;
assert!(handshake.starts_with("GET /v1/protocol/ws "));
assert!(
handshake
.to_ascii_lowercase()
.contains("authorization: bearer ")
);
observed(fixture.executor.drain()).await.unwrap();
let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone());
let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone());
let server = fixture.server.clone();
@@ -1546,9 +1640,17 @@ async fn cancelled_runtime_key_revoke_commits_audit_and_removes_broker_registrat
"/api/w/db-execution/runtimes/revoke-runtime/trust-key",
serde_json::json!({"expected_binding_id": binding.binding_id}),
);
let revoke =
let mut revoke =
tokio::spawn(async move { dispatch_workspace_request(State(server), request).await });
observed(waiting).await.unwrap(); // Actual authenticated revoke reached BEGIN IMMEDIATE.
tokio::select! {
reached = observed(waiting) => reached.unwrap(), // Actual authenticated BEGIN IMMEDIATE.
response = &mut revoke => {
let response = response.unwrap();
let status = response.status();
let body = axum::body::to_bytes(response.into_body(), 65536).await.unwrap();
panic!("revoke returned before its SQLite wait: {status}: {}", String::from_utf8_lossy(&body));
}
}
runtime_progress_while_locked(&lock).await;
revoke.abort();
assert!(revoke.await.unwrap_err().is_cancelled());
crates/workspace-server/src/server_drive_grants.rs
+10 -4
View File
@@ -64,8 +64,9 @@ pub(super) async fn authorized_drive(
// incomplete, or inconsistent with its transport identity.
let headers = current_worker_contract_headers(context)
.map_err(|_| fail(DriveApiErrorCode::Denied))?;
let worker = current_worker_identity(api, workspace_id, &headers)
.map_err(|_| fail(DriveApiErrorCode::Denied))?;
let worker = authenticate_worker_mutation_source_owned(api, workspace_id, &headers)
.await
.map_err(pre_error)?;
if context.worker_source.as_ref().is_some_and(|source| {
source.runtime_id != worker.runtime_id || source.worker_id != worker.worker_id
}) {
@@ -139,9 +140,14 @@ pub(super) async fn create_grant(
}
}
let worker = RuntimeWorkerRef::new(&request.runtime_id, &request.worker_id);
let operation_api = api.clone();
let summary = api
.runtime
.worker(&worker)
.host_executors
.host()
.map_err(pre_error)?
.run(move || Ok::<_, Error>(operation_api.runtime.worker(&worker)))
.await
.map_err(pre_error)?
.map_err(|_| fail(DriveApiErrorCode::NotFound))?;
if summary.workspace.workspace_id.as_deref() != Some(workspace_id) {
return Err(fail(DriveApiErrorCode::Denied));
crates/workspace-server/src/server_residual_execution_tests.rs
+478 -1
View File
@@ -561,7 +561,14 @@ async fn ticket_source_ref_reads_and_ready_publication_leave_sql_and_tokio_free(
},
));
assert!(futures::poll!(&mut response).is_pending());
let request = observed(reply.entered.take().unwrap()).await.unwrap();
// Ready reaches the source POST only after mutation and best-effort
// notification. Bound this multi-stage setup with the fixture watchdog,
// not the shorter budget used below to prove timer/independent control
// progress while the actual provider request is held.
let request = tokio::select! {
request = tokio::time::timeout(WATCHDOG, reply.entered.take().unwrap()) => request.expect("source observation setup did not reach the provider").unwrap(),
result = &mut response => panic!("{operation} returned before the held source observation: {result:?}"),
};
assert!(
request.contains("/v1/repository-refs/observe"),
"{operation}: {request}"
@@ -902,3 +909,473 @@ async fn cancelled_feature_shutdown_keeps_sql_close_owned_without_blocking_contr
"accepted shutdown must fence later use"
);
}
#[tokio::test(flavor = "current_thread")]
async fn session_workdir_and_upload_callers_keep_provider_waits_off_sql_and_tokio() {
use server_api::ServerApi;
for operation in ["session", "history", "attachment", "workdirs", "upload"] {
let fixture = fixture().await;
let mut reply = HeldRuntimeReply::new();
install_remote_fixture(
&fixture,
"caller-runtime",
&format!("http://{}", reply.address),
false,
);
let worker_id = Uuid::now_v7().to_string();
fixture
.store
.upsert_worker_registry(&registry_record(RuntimeWorkerRef::new(
"caller-runtime",
&worker_id,
)))
.unwrap();
let actor = RequestActor {
user_id: "fixture-user".into(),
account_id: format!("account-{WORKSPACE}"),
handle: "fixture-user".into(),
display_name: "Fixture user".into(),
auth_method: ActorAuthMethod::BrowserSession,
};
fixture.api.attachment_upload_grants.lock().unwrap().insert(
"caller-upload".into(),
AttachmentUploadGrant {
workspace_id: WORKSPACE.into(),
runtime_id: "caller-runtime".into(),
worker_id: worker_id.clone(),
account_id: actor.account_id.clone(),
file_name: "notes.txt".into(),
media_type: "text/plain".into(),
expires_at_ms: u64::MAX,
state: AttachmentUploadGrantState::Pending,
},
);
let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let api = fixture.api.clone();
let task = tokio::spawn(crate::db_executor::with_operation_lifetime(
Arc::new(CountJobLifetime(leases.clone())),
async move {
let service = ServerApiContractService::Workspace(api.clone());
match operation {
"session" => {
let _ = service
.worker_session(WORKSPACE.into(), "caller-runtime".into(), worker_id)
.await;
}
"history" => {
let _ = service
.worker_session_history(
WORKSPACE.into(),
"caller-runtime".into(),
worker_id,
server_api::WorkspaceWorkerSessionHistoryQuery {
cursor: None,
limit: Some(1),
},
)
.await;
}
"attachment" => {
let _ = scoped_worker_session_attachment(
State(api),
AxumPath((
WORKSPACE.into(),
"caller-runtime".into(),
worker_id,
"session".into(),
"attachment".into(),
)),
Extension(actor),
)
.await;
}
"workdirs" => {
let _ = scoped_list_working_directories(
State(api),
AxumPath(ScopedWorkspacePath {
workspace_id: WORKSPACE.into(),
}),
)
.await;
}
"upload" => {
let _ = upload_runtime_worker_file_contract(
api,
actor,
ScopedAttachmentUploadPath {
workspace_id: WORKSPACE.into(),
runtime_id: "caller-runtime".into(),
worker_id,
upload_id: "caller-upload".into(),
},
b"notes",
)
.await;
}
_ => unreachable!(),
}
},
));
let request = observed(reply.entered.take().unwrap()).await.unwrap();
assert!(request.contains("/v1/"), "{operation}: {request}");
assert!(
!reply.upstream_released.load(Ordering::SeqCst),
"{operation}: watchdog release is not executor progress"
);
let store = fixture.store.clone();
observed(
fixture
.executor
.run(move || store.get_workspace_memory_settings(WORKSPACE)),
)
.await
.unwrap();
assert_eq!(
observed(tokio::spawn(async { "control" })).await.unwrap(),
"control"
);
observed(tokio::time::sleep(StdDuration::from_millis(1))).await;
let mut queued = Box::pin(fixture.host_executor.run(|| Ok::<_, Error>(())));
assert!(futures::poll!(&mut queued).is_pending());
let error = scoped_list_working_directories(
State(fixture.api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: WORKSPACE.into(),
}),
)
.await
.unwrap_err();
response_json(error.into_response(), StatusCode::SERVICE_UNAVAILABLE).await;
task.abort();
assert!(task.await.unwrap_err().is_cancelled());
assert_eq!(
leases.load(Ordering::SeqCst),
1,
"{operation}: lost provider owner"
);
reply.finish();
observed(queued).await.unwrap();
observed(fixture.host_executor.drain()).await.unwrap();
observed(fixture.executor.drain()).await.unwrap();
assert_eq!(
leases.load(Ordering::SeqCst),
0,
"{operation}: retained owner after completion"
);
if operation == "upload" {
assert!(
matches!(
fixture
.api
.attachment_upload_grants
.lock()
.unwrap()
.get("caller-upload")
.unwrap()
.state,
AttachmentUploadGrantState::Pending
),
"failed Runtime lookup must finish the accepted grant claim even with a cancelled waiter"
);
}
}
}
#[tokio::test(flavor = "current_thread")]
async fn repository_create_keeps_cancelled_sql_commit_owned_and_replays_one_intent() {
let fixture = fixture().await;
let (mut busy, waiting) = BusyObservation::install_store(fixture.api.config_store.clone());
let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone());
let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let request = || {
Fixture::request(
"POST",
&format!("/api/w/{WORKSPACE}/repositories"),
serde_json::json!({ "repository_key": "owned-repository", "source": "https://example.invalid/repo.git", "default_ref": "main" }),
)
};
let mut response = Box::pin(crate::db_executor::with_operation_lifetime(
Arc::new(CountJobLifetime(leases.clone())),
fixture.inner_router().oneshot(request()),
));
tokio::select! {
result = &mut response => panic!("repository must wait for its write transaction: {:?}", result.unwrap().status()),
reached = observed(waiting) => reached.unwrap(),
}
runtime_progress_while_locked(&lock).await;
assert_eq!(
observed(fixture.host_executor.run(|| Ok::<_, Error>("control")))
.await
.unwrap(),
"control"
);
drop(response);
assert_eq!(leases.load(Ordering::SeqCst), 1);
lock.release();
busy.resume();
observed(fixture.executor.drain()).await.unwrap();
assert_eq!(leases.load(Ordering::SeqCst), 0);
let rows = fixture.store.list_repositories(WORKSPACE).unwrap();
assert_eq!(
rows.iter()
.filter(|record| record.repository_key == "owned-repository")
.count(),
1
);
let replay = observed(fixture.inner_router().oneshot(request()))
.await
.unwrap();
let body = response_json(replay, StatusCode::OK).await;
assert_eq!(body["replayed"], true);
assert_eq!(
fixture.store.list_repositories(WORKSPACE).unwrap().len(),
rows.len()
);
}
#[tokio::test(flavor = "current_thread")]
async fn drive_and_config_preflight_keep_live_authority_owned_and_report_overload_as_availability()
{
async fn preflight(
api: &WorkspaceApi,
context: &server_api::ServerRequestContext,
operation: &str,
) -> u16 {
match operation {
"config" => {
workspace_config::identity(api, context, WORKSPACE)
.await
.err()
.unwrap()
.status
}
"drive" => api_macros::HttpError::status_code(
&drive_grants::authorized_drive(api, context, WORKSPACE, false)
.await
.err()
.unwrap(),
),
"grant" => api_macros::HttpError::status_code(
&drive_grants::create_grant(
api,
context,
WORKSPACE,
server_api::DriveGrantCreateRequest {
runtime_id: "preflight-runtime".into(),
worker_id: "target-worker".into(),
access: server_api::DriveAccess::ReadOnly,
},
)
.await
.err()
.unwrap(),
),
_ => unreachable!(),
}
}
for operation in ["config", "drive", "grant"] {
let fixture = fixture().await;
let mut reply = HeldRuntimeReply::new();
install_remote_fixture(
&fixture,
"preflight-runtime",
&format!("http://{}", reply.address),
false,
);
let context = server_api::ServerRequestContext {
actor: if operation == "grant" {
Some(RequestActor {
user_id: "fixture-user".into(),
account_id: format!("account-{WORKSPACE}"),
handle: "fixture-user".into(),
display_name: "Fixture user".into(),
auth_method: ActorAuthMethod::BrowserSession,
})
} else {
None
},
worker_source: None,
runtime_source: if operation == "grant" {
None
} else {
Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: "preflight-runtime".into(),
worker_id: Some("source-worker".into()),
})
},
origin: None,
transport_headers: Vec::new(),
};
let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let api = fixture.api.clone();
let request_context = context.clone();
let task = tokio::spawn(crate::db_executor::with_operation_lifetime(
Arc::new(CountJobLifetime(leases.clone())),
async move { preflight(&api, &request_context, operation).await },
));
let request = observed(reply.entered.take().unwrap()).await.unwrap();
assert!(request.contains("/v1/workers/"), "{operation}: {request}");
let store = fixture.store.clone();
observed(
fixture
.executor
.run(move || store.get_workspace_memory_settings(WORKSPACE)),
)
.await
.unwrap();
observed(tokio::time::sleep(StdDuration::from_millis(1))).await;
assert_eq!(
observed(tokio::spawn(async { "control" })).await.unwrap(),
"control"
);
assert!(!reply.upstream_released.load(Ordering::SeqCst));
task.abort();
assert!(task.await.unwrap_err().is_cancelled());
assert_eq!(leases.load(Ordering::SeqCst), 1);
fixture.host_executor.dispatch(|| {}).unwrap();
assert_eq!(
observed(preflight(&fixture.api, &context, operation)).await,
503,
"{operation}: overload must not be authority denial"
);
reply.finish();
observed(fixture.host_executor.drain()).await.unwrap();
observed(fixture.executor.drain()).await.unwrap();
assert_eq!(leases.load(Ordering::SeqCst), 0);
}
}
#[tokio::test(flavor = "current_thread")]
async fn ui_config_commit_publishes_after_commit_without_holding_the_sql_lane() {
let fixture = fixture().await;
let current = fixture
.api
.config_store
.load_workspace_config(WORKSPACE)
.unwrap()
.unwrap();
let mut reply = HeldRuntimeReply::new();
install_remote_fixture(
&fixture,
"prompt-runtime",
&format!("http://{}", reply.address),
false,
);
let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let mut commit = Box::pin(crate::db_executor::with_operation_lifetime(
Arc::new(CountJobLifetime(leases.clone())),
scoped_commit_workspace_config_tree(
State(fixture.api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: WORKSPACE.into(),
}),
Json(server_api::ConfigCommitRequest {
base_digest: current.snapshot.digest.clone(),
entrypoints: current
.contract
.entrypoints
.iter()
.map(ToString::to_string)
.collect(),
changes: vec![server_api::ConfigTreeChange::Create {
path: "notes/owned-publication.txt".into(),
content_type: server_api::ConfigContentType::Text,
content: "committed before prompt publication".into(),
}],
}),
),
));
assert!(futures::poll!(&mut commit).is_pending());
let request = observed(reply.entered.take().unwrap()).await.unwrap();
assert!(
request.starts_with("POST "),
"prompt publication: {request}"
);
let config_store = fixture.api.config_store.clone();
let state = observed(
fixture
.executor
.run(move || config_store.load_workspace_config(WORKSPACE)),
)
.await
.unwrap()
.unwrap();
assert_ne!(
state.snapshot.digest, current.snapshot.digest,
"publication must observe the committed tree"
);
assert!(
state
.snapshot
.entries
.keys()
.any(|path| path.to_string() == "notes/owned-publication.txt")
);
observed(tokio::time::sleep(StdDuration::from_millis(1))).await;
assert_eq!(
observed(tokio::spawn(async { "control" })).await.unwrap(),
"control"
);
assert!(!reply.upstream_released.load(Ordering::SeqCst));
drop(commit);
assert_eq!(leases.load(Ordering::SeqCst), 1);
reply.finish();
observed(fixture.host_executor.drain()).await.unwrap();
observed(fixture.executor.drain()).await.unwrap();
assert_eq!(leases.load(Ordering::SeqCst), 0);
assert_eq!(
fixture
.api
.config_store
.load_workspace_config(WORKSPACE)
.unwrap()
.unwrap()
.snapshot
.digest,
state.snapshot.digest,
"failed best-effort publication must not repeat or roll back the commit"
);
}
#[tokio::test(flavor = "current_thread")]
async fn repository_provider_reads_do_not_consume_the_sql_execution_queue() {
let fixture = fixture().await;
let (entered, waiting) = oneshot::channel();
let (release, permit) = mpsc::channel();
fixture
.executor
.dispatch(move || {
let _ = entered.send(());
let _ = permit.recv_timeout(WATCHDOG);
})
.unwrap();
observed(waiting).await.unwrap();
fixture.executor.dispatch(|| {}).unwrap();
let probe = fixture.executor.run(|| Ok::<_, Error>(())).await;
assert!(matches!(probe, Err(Error::DbCapacityExceeded)));
// Empty catalog / absent key are deliberate boundary inputs: provider
// reads keep their ordinary response even when SQL job admission is full.
let list = observed(list_repositories(State(fixture.api.clone())))
.await
.unwrap();
assert!(list.0.items.is_empty());
let detail = observed(repository_detail(
State(fixture.api.clone()),
AxumPath("missing".into()),
))
.await
.unwrap_err();
response_json(detail.into_response(), StatusCode::NOT_FOUND).await;
let log = observed(repository_log(
State(fixture.api.clone()),
AxumPath("missing".into()),
Query(LogQuery { limit: Some(1) }),
))
.await
.unwrap_err();
response_json(log.into_response(), StatusCode::NOT_FOUND).await;
let _ = release.send(());
observed(fixture.executor.drain()).await.unwrap();
observed(fixture.host_executor.drain()).await.unwrap();
}
crates/workspace-server/src/server_workspace_config.rs
+4 -2
View File
@@ -62,7 +62,7 @@ fn unknown() -> Failure {
..fail(500, "outcome_unknown")
}
}
pub(super) fn identity(
pub(super) async fn identity(
api: &WorkspaceApi,
context: &server_api::ServerRequestContext,
workspace_id: &str,
@@ -70,7 +70,9 @@ pub(super) fn identity(
validate_workspace_scope(api, workspace_id).map_err(pre_error)?;
let headers =
current_worker_contract_headers(context).map_err(|_| fail(403, "access_denied"))?;
current_worker_identity(api, workspace_id, &headers).map_err(pre_error)
authenticate_worker_mutation_source_owned(api, workspace_id, &headers)
.await
.map_err(pre_error)
}
async fn lock(
api: &WorkspaceApi,