fix: restore sidebar workdir projection
This commit is contained in:
@@ -8529,6 +8529,9 @@ async fn scoped_attach_current_worker_workdir(
|
|||||||
);
|
);
|
||||||
return Err(error.into());
|
return Err(error.into());
|
||||||
}
|
}
|
||||||
|
api.worker_projection
|
||||||
|
.publish_catalog_change(&worker)
|
||||||
|
.map_err(ApiError::from)?;
|
||||||
Ok(Json(CurrentWorkerWorkdirAttachmentResponse {
|
Ok(Json(CurrentWorkerWorkdirAttachmentResponse {
|
||||||
workspace_id: api.config.workspace_id.clone(),
|
workspace_id: api.config.workspace_id.clone(),
|
||||||
workdir_id: workdir_id.to_string(),
|
workdir_id: workdir_id.to_string(),
|
||||||
@@ -8553,6 +8556,9 @@ async fn scoped_detach_current_worker_workdir(
|
|||||||
Some(&link.workdir_id),
|
Some(&link.workdir_id),
|
||||||
&Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
|
&Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
|
||||||
)?;
|
)?;
|
||||||
|
api.worker_projection
|
||||||
|
.publish_catalog_change(&worker)
|
||||||
|
.map_err(ApiError::from)?;
|
||||||
Ok(Json(CurrentWorkerWorkdirAttachmentResponse {
|
Ok(Json(CurrentWorkerWorkdirAttachmentResponse {
|
||||||
workspace_id: api.config.workspace_id.clone(),
|
workspace_id: api.config.workspace_id.clone(),
|
||||||
workdir_id: link.workdir_id,
|
workdir_id: link.workdir_id,
|
||||||
@@ -18126,6 +18132,9 @@ fn link_worker_to_workdir(
|
|||||||
} else {
|
} else {
|
||||||
api.store.attach_worker_workdir(&record)?;
|
api.store.attach_worker_workdir(&record)?;
|
||||||
}
|
}
|
||||||
|
api.worker_projection
|
||||||
|
.publish_catalog_change(&worker_record.worker)
|
||||||
|
.map_err(ApiError::from)?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -31982,9 +31991,52 @@ mod tests {
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
assert_eq!(spawned.state, WorkerOperationState::Accepted);
|
assert_eq!(spawned.state, WorkerOperationState::Accepted);
|
||||||
let worker_id = spawned.worker.unwrap().worker.worker_id;
|
let worker_id = spawned.worker.unwrap().worker.worker_id;
|
||||||
|
let (_, projections) = api
|
||||||
|
.store
|
||||||
|
.worker_registry_projection_snapshot(TEST_WORKSPACE_ID, 10)
|
||||||
|
.unwrap();
|
||||||
|
let observation = projections
|
||||||
|
.iter()
|
||||||
|
.find(|projection| projection.registry.worker.worker_id == worker_id)
|
||||||
|
.and_then(|projection| projection.observation.as_ref())
|
||||||
|
.expect("spawned Worker observation exists");
|
||||||
|
assert!(observation.worker.repository_id.is_none());
|
||||||
|
assert!(observation.worker.working_directory_id.is_none());
|
||||||
|
|
||||||
|
let workdir_id = "001a0b415233500000c";
|
||||||
|
api.store
|
||||||
|
.upsert_workdir_registry(&WorkdirRegistryRecord {
|
||||||
|
workspace_id: TEST_WORKSPACE_ID.to_string(),
|
||||||
|
workdir_id: workdir_id.to_string(),
|
||||||
|
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
|
||||||
|
repository_id: test_repository_id(&api),
|
||||||
|
creation_selector: Some("HEAD".to_string()),
|
||||||
|
creation_ref: None,
|
||||||
|
creation_tree: None,
|
||||||
|
current_selector: Some("work/sidebar-projection".to_string()),
|
||||||
|
current_ref: None,
|
||||||
|
current_tree: None,
|
||||||
|
observed_at_epoch_seconds: None,
|
||||||
|
materialization_status: "present".to_string(),
|
||||||
|
cleanliness: "clean".to_string(),
|
||||||
|
created_at: "1".to_string(),
|
||||||
|
updated_at: "1".to_string(),
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
api.store
|
||||||
|
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
|
||||||
|
workspace_id: TEST_WORKSPACE_ID.to_string(),
|
||||||
|
worker: RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, &worker_id),
|
||||||
|
workdir_id: workdir_id.to_string(),
|
||||||
|
role: "attachment".to_string(),
|
||||||
|
linked_at: "1".to_string(),
|
||||||
|
unlinked_at: None,
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||||
let address = listener.local_addr().unwrap();
|
let address = listener.local_addr().unwrap();
|
||||||
|
let projection_api = api.clone();
|
||||||
let app = build_inner_router(api).layer(Extension(test_browser_request_actor()));
|
let app = build_inner_router(api).layer(Extension(test_browser_request_actor()));
|
||||||
let server = tokio::spawn(async move {
|
let server = tokio::spawn(async move {
|
||||||
let _ = axum::serve(listener, app).await;
|
let _ = axum::serve(listener, app).await;
|
||||||
@@ -32022,13 +32074,57 @@ mod tests {
|
|||||||
) => workers,
|
) => workers,
|
||||||
other => panic!("expected Workspace Worker snapshot, got {other:?}"),
|
other => panic!("expected Workspace Worker snapshot, got {other:?}"),
|
||||||
};
|
};
|
||||||
|
let projected_worker = workers
|
||||||
|
.iter()
|
||||||
|
.find(|worker| worker.worker_id.as_str() == worker_id)
|
||||||
|
.expect("spawned Worker is projected");
|
||||||
|
assert_eq!(projected_worker.resource_key.as_deref(), Some("W-1"));
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
workers
|
projected_worker.repository_key.as_deref(),
|
||||||
.iter()
|
Some("test-repository")
|
||||||
.find(|worker| worker.worker_id.as_str() == worker_id)
|
|
||||||
.and_then(|worker| worker.resource_key.as_deref()),
|
|
||||||
Some("W-1")
|
|
||||||
);
|
);
|
||||||
|
assert_eq!(
|
||||||
|
projected_worker
|
||||||
|
.working_directory_id
|
||||||
|
.as_ref()
|
||||||
|
.map(|workdir_id| workdir_id.as_str()),
|
||||||
|
Some(workdir_id)
|
||||||
|
);
|
||||||
|
|
||||||
|
projection_api
|
||||||
|
.store
|
||||||
|
.detach_worker_workdir(
|
||||||
|
TEST_WORKSPACE_ID,
|
||||||
|
&RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, &worker_id),
|
||||||
|
Some(workdir_id),
|
||||||
|
"2",
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
projection_api
|
||||||
|
.worker_projection
|
||||||
|
.publish_catalog_change(&RuntimeWorkerRef::new(
|
||||||
|
EMBEDDED_WORKER_RUNTIME_ID,
|
||||||
|
&worker_id,
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
let Message::Text(text) = socket.next().await.unwrap().unwrap() else {
|
||||||
|
panic!("expected Worker projection update");
|
||||||
|
};
|
||||||
|
let update: protocol::subscription::SubscriptionFrame =
|
||||||
|
serde_json::from_str(text.as_str()).unwrap();
|
||||||
|
let detached_worker = match update.payload {
|
||||||
|
protocol::subscription::SubscriptionFramePayload::Event(
|
||||||
|
protocol::subscription::SubscriptionEvent::Event {
|
||||||
|
payload:
|
||||||
|
protocol::subscription::SubscriptionEventPayload::WorkerUpserted { worker },
|
||||||
|
..
|
||||||
|
},
|
||||||
|
) => worker,
|
||||||
|
other => panic!("expected Worker upsert after Workdir detach, got {other:?}"),
|
||||||
|
};
|
||||||
|
assert_eq!(detached_worker.worker_id.as_str(), worker_id);
|
||||||
|
assert!(detached_worker.repository_key.is_none());
|
||||||
|
assert!(detached_worker.working_directory_id.is_none());
|
||||||
|
|
||||||
let subscribe_protocol = protocol::subscription::SubscriptionFrame::new(
|
let subscribe_protocol = protocol::subscription::SubscriptionFrame::new(
|
||||||
protocol::subscription::SubscriptionFramePayload::Request(
|
protocol::subscription::SubscriptionFramePayload::Request(
|
||||||
|
|||||||
@@ -423,12 +423,56 @@ fn project_registry_worker(
|
|||||||
worker.workspace_id = Some(record.registry.workspace_id);
|
worker.workspace_id = Some(record.registry.workspace_id);
|
||||||
worker.display_name = Some(record.registry.display_name);
|
worker.display_name = Some(record.registry.display_name);
|
||||||
worker.profile = record.registry.profile;
|
worker.profile = record.registry.profile;
|
||||||
if !project_repository_key(api, &mut worker) {
|
if !project_working_directory(api, &record.registry.worker, &mut worker) {
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
Some(worker)
|
Some(worker)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn project_working_directory(
|
||||||
|
api: &WorkspaceApi,
|
||||||
|
worker_ref: &RuntimeWorkerRef,
|
||||||
|
worker: &mut SubscriptionWorker,
|
||||||
|
) -> bool {
|
||||||
|
let Ok(links) = api
|
||||||
|
.store
|
||||||
|
.list_worker_workdir_links(&api.config.workspace_id, worker_ref)
|
||||||
|
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;
|
||||||
|
};
|
||||||
|
if workdir.runtime_id != worker_ref.runtime_id {
|
||||||
|
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);
|
||||||
|
} else {
|
||||||
|
let Ok(has_link_history) = api
|
||||||
|
.store
|
||||||
|
.worker_workdir_link_history_exists(&api.config.workspace_id, worker_ref)
|
||||||
|
else {
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
if has_link_history {
|
||||||
|
worker.repository_id = None;
|
||||||
|
worker.repository_key = None;
|
||||||
|
worker.working_directory_id = None;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
project_repository_key(api, worker)
|
||||||
|
}
|
||||||
|
|
||||||
fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
|
fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
|
||||||
let Some(repository_id) = worker.repository_id.take() else {
|
let Some(repository_id) = worker.repository_id.take() else {
|
||||||
worker.repository_key = None;
|
worker.repository_key = None;
|
||||||
|
|||||||
Reference in New Issue
Block a user