michel/clusterflux-public
1
0
Fork 0
Code Issues Pull requests Projects Releases 21 Packages Wiki Activity Actions

Export artifact bytes explicitly

Browse source
This commit is contained in:
Michel Paulissen 2026-07-03 23:43:45 +02:00
parent 2a86900e57
commit 22fa4fc675
9 changed files with 346 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

1
crates/disasmer-cli/Cargo.toml
View file

@ -11,6 +11,7 @@ path = "src/main.rs"
[dependencies]
anyhow.workspace = true
base64.workspace = true
clap.workspace = true
disasmer-core = { path = "../disasmer-core" }
serde.workspace = true

228
crates/disasmer-cli/src/main.rs
View file

@ -6,6 +6,7 @@ use std::process::{Command, Stdio};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result};
use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _};
use clap::{Args, Parser, Subcommand};
use disasmer_core::{
diagnose_environment_references, BrowserLoginFlow, BundleIdentityInputs, BundleMetadata,
@ -22,6 +23,7 @@ const BROWSER_CALLBACK_ADDR: &str = "127.0.0.1:45173";
const BROWSER_CALLBACK_PATH: &str = "/callback";
const DEFAULT_BROWSER_LOGIN_CALLBACK_TIMEOUT_SECONDS: u64 = 300;
const DOCTOR_COORDINATOR_TIMEOUT: Duration = Duration::from_millis(500);
const DEFAULT_ARTIFACT_EXPORT_MAX_BYTES: u64 = 64 * 1024 * 1024;
#[derive(Clone, Debug, Parser)]
#[command(name = "disasmer", version, arg_required_else_help = true)]
@ -1524,7 +1526,19 @@ fn artifact_export_report(args: ArtifactExportArgs) -> Result<Value> {
"direct_connectivity": true,
"failure_reason": "",
}))?;
let export_plan = artifact_export_plan_summary(&response, &args.to);
let mut export_plan = artifact_export_plan_summary(&response, &args.to);
let local_export =
if response.get("type").and_then(Value::as_str) == Some("artifact_export_plan") {
artifact_export_local_write_followup(&mut session, &args, &response)?
} else {
json!({
"status": "skipped",
"explicit_user_action": true,
"local_path": &args.to,
"local_bytes_written_by_cli": false,
})
};
apply_local_export_summary(&mut export_plan, &local_export);
return Ok(json!({
"command": "artifact export",
"coordinator": coordinator,
@ -1532,6 +1546,7 @@ fn artifact_export_report(args: ArtifactExportArgs) -> Result<Value> {
"to": args.to,
"receiver_node": args.receiver_node,
"export_plan": export_plan,
"local_export": local_export,
"response": response,
"coordinator_session_requests": session.requests(),
}));
@ -1551,6 +1566,145 @@ fn artifact_export_report(args: ArtifactExportArgs) -> Result<Value> {
}))
}
fn artifact_export_local_write_followup(
session: &mut JsonLineSession,
args: &ArtifactExportArgs,
export_response: &Value,
) -> Result<Value> {
let artifact_size = export_response
.get("artifact_size_bytes")
.and_then(Value::as_u64)
.unwrap_or(DEFAULT_ARTIFACT_EXPORT_MAX_BYTES);
let nonce = command_nonce("artifact-export");
let link_response = session.request(json!({
"type": "create_artifact_download_link",
"tenant": args.scope.tenant,
"project": args.scope.project,
"actor_user": args.scope.user,
"artifact": args.artifact,
"max_bytes": DEFAULT_ARTIFACT_EXPORT_MAX_BYTES,
"token_nonce": nonce,
"now_epoch_seconds": 0,
}))?;
if link_response.get("type").and_then(Value::as_str) != Some("artifact_download_link") {
return Ok(json!({
"status": "download_link_failed",
"explicit_user_action": true,
"local_path": &args.to,
"local_bytes_written_by_cli": false,
"content_bytes_available": false,
"link_response": link_response,
}));
}
let token_digest = link_response
.pointer("/link/scoped_token_digest")
.cloned()
.context("artifact download link did not include a scoped token digest")?;
let stream_response = session.request(json!({
"type": "open_artifact_download_stream",
"tenant": args.scope.tenant,
"project": args.scope.project,
"actor_user": args.scope.user,
"artifact": args.artifact,
"max_bytes": DEFAULT_ARTIFACT_EXPORT_MAX_BYTES,
"token_nonce": nonce,
"token_digest": token_digest,
"now_epoch_seconds": 1,
"chunk_bytes": artifact_size,
}))?;
if stream_response.get("type").and_then(Value::as_str) != Some("artifact_download_stream") {
return Ok(json!({
"status": "download_stream_failed",
"explicit_user_action": true,
"local_path": &args.to,
"local_bytes_written_by_cli": false,
"content_bytes_available": false,
"stream": artifact_stream_summary(&stream_response),
}));
}
let Some(content_base64) = stream_response
.get("content_base64")
.and_then(Value::as_str)
else {
return Ok(json!({
"status": "download_stream_opened_without_content",
"explicit_user_action": true,
"local_path": &args.to,
"local_bytes_written_by_cli": false,
"content_bytes_available": false,
"stream": artifact_stream_summary(&stream_response),
}));
};
let bytes = BASE64_STANDARD
.decode(content_base64)
.context("artifact download stream returned invalid base64 content")?;
if let Some(parent) = args.to.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
std::fs::write(&args.to, &bytes)
.with_context(|| format!("failed to write {}", args.to.display()))?;
Ok(json!({
"status": "local_bytes_written",
"explicit_user_action": true,
"local_path": &args.to,
"local_bytes_written_by_cli": true,
"bytes_written": bytes.len(),
"content_bytes_available": true,
"content_source": stream_response.get("content_source").cloned().unwrap_or(Value::Null),
"stream": artifact_stream_summary(&stream_response),
}))
}
fn artifact_stream_summary(response: &Value) -> Value {
json!({
"status": response.get("type").and_then(Value::as_str).unwrap_or("coordinator_response"),
"streamed_bytes": response.get("streamed_bytes").cloned().unwrap_or_else(|| json!(0)),
"charged_download_bytes": response.get("charged_download_bytes").cloned().unwrap_or_else(|| json!(0)),
"content_bytes_available": response.get("content_bytes_available").cloned().unwrap_or_else(|| json!(false)),
"content_source": response.get("content_source").cloned().unwrap_or(Value::Null),
"content_material_returned_in_report": false,
"error": response.get("message").cloned().unwrap_or(Value::Null),
})
}
fn apply_local_export_summary(export_plan: &mut Value, local_export: &Value) {
let Some(object) = export_plan.as_object_mut() else {
return;
};
object.insert(
"local_bytes_written_by_cli".to_owned(),
local_export
.get("local_bytes_written_by_cli")
.cloned()
.unwrap_or_else(|| json!(false)),
);
object.insert(
"local_export_status".to_owned(),
local_export
.get("status")
.cloned()
.unwrap_or_else(|| json!("unknown")),
);
object.insert(
"content_bytes_available".to_owned(),
local_export
.get("content_bytes_available")
.cloned()
.unwrap_or_else(|| json!(false)),
);
if let Some(bytes_written) = local_export.get("bytes_written") {
object.insert("bytes_written".to_owned(), bytes_written.clone());
object.insert(
"writes_require_data_plane_followup".to_owned(),
json!(false),
);
}
}
fn dap_plan(args: DapArgs) -> Result<Value> {
Ok(json!({
"command": "dap",
@ -3273,6 +3427,7 @@ fn artifact_export_plan_summary(response: &Value, to: &Path) -> Value {
"local_bytes_written_by_cli": false,
"writes_require_data_plane_followup": true,
"default_durable_store_assumed": false,
"artifact_size_bytes": response.get("artifact_size_bytes").cloned().unwrap_or(Value::Null),
"source_node": response.get("source_node").cloned().unwrap_or(Value::Null),
"receiver_node": response.get("receiver_node").cloned().unwrap_or(Value::Null),
"transport": plan.get("transport").cloned().unwrap_or(Value::Null),
@ -6364,11 +6519,33 @@ mod tests {
r#"{"type":"artifact_download_link","link":{"artifact":"app.txt","source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":2048,"policy_context_digest":"sha256:policy"}}"#
);
let export_response = concat!(
r#"{"type":"artifact_export_plan","source_node":"node-a","receiver_node":"node-b","plan":{"transport":"NativeQuic","scope":{"tenant":"tenant","project":"project","process":"vp","object":{"Artifact":"app.txt"},"authorization_subject":"artifact-export:node-a-to-node-b"},"source":{"node":"node-a","advertised_addr":"quic://node-a","public_key_fingerprint":"sha256:source"},"destination":{"node":"node-b","advertised_addr":"quic://node-b","public_key_fingerprint":"sha256:destination"},"authorization_digest":"sha256:auth","coordinator_assisted_rendezvous":true,"coordinator_bulk_relay_allowed":false}}"#
r#"{"type":"artifact_export_plan","source_node":"node-a","receiver_node":"node-b","artifact_size_bytes":9,"plan":{"transport":"NativeQuic","scope":{"tenant":"tenant","project":"project","process":"vp","object":{"Artifact":"app.txt"},"authorization_subject":"artifact-export:node-a-to-node-b"},"source":{"node":"node-a","advertised_addr":"quic://node-a","public_key_fingerprint":"sha256:source"},"destination":{"node":"node-b","advertised_addr":"quic://node-b","public_key_fingerprint":"sha256:destination"},"authorization_digest":"sha256:auth","coordinator_assisted_rendezvous":true,"coordinator_bulk_relay_allowed":false}}"#
);
for expected in ["create_artifact_download_link", "export_artifact_to_node"] {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let export_download_response = concat!(
r#"{"type":"artifact_download_link","link":{"artifact":"app.txt","source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:export-token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":67108864,"policy_context_digest":"sha256:policy"}}"#
);
let export_stream_response = concat!(
r#"{"type":"artifact_download_stream","link":{"artifact":"app.txt","source":{"RetainedNode":"node-a"},"url_path":"/artifacts/tenant/project/vp/app.txt","scoped_token_digest":"sha256:export-token","expires_at_epoch_seconds":60,"tenant":"tenant","project":"project","process":"vp","actor":{"User":"user"},"max_bytes":67108864,"policy_context_digest":"sha256:policy"},"streamed_bytes":9,"charged_download_bytes":9,"content_bytes_available":true,"content_base64":"YXBwLWJ5dGVz","content_source":"coordinator_complete_stdout_tail"}"#
);
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(r#""type":"create_artifact_download_link""#));
assert!(line.contains(r#""tenant":"tenant""#));
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""artifact":"app.txt""#));
assert!(line.contains(r#""max_bytes":2048"#));
stream.write_all(download_response.as_bytes()).unwrap();
stream.write_all(b"\n").unwrap();
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
for (expected, phase) in [
("export_artifact_to_node", "export-plan"),
("create_artifact_download_link", "export-download"),
("open_artifact_download_stream", "export-stream"),
] {
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.contains(&format!(r#""type":"{expected}""#)));
@ -6376,11 +6553,21 @@ mod tests {
assert!(line.contains(r#""project":"project""#));
assert!(line.contains(r#""artifact":"app.txt""#));
if expected == "create_artifact_download_link" {
assert!(line.contains(r#""max_bytes":2048"#));
stream.write_all(download_response.as_bytes()).unwrap();
} else {
assert_eq!(phase, "export-download");
assert!(line.contains(&format!(
r#""max_bytes":{}"#,
DEFAULT_ARTIFACT_EXPORT_MAX_BYTES
)));
stream
.write_all(export_download_response.as_bytes())
.unwrap();
} else if expected == "export_artifact_to_node" {
assert!(line.contains(r#""receiver_node":"node-b""#));
stream.write_all(export_response.as_bytes()).unwrap();
} else {
assert!(line.contains(r#""chunk_bytes":9"#));
assert!(line.contains(r#""token_digest":"sha256:export-token""#));
stream.write_all(export_stream_response.as_bytes()).unwrap();
}
stream.write_all(b"\n").unwrap();
}
@ -6393,6 +6580,8 @@ mod tests {
json: false,
};
let temp = tempfile::tempdir().unwrap();
let export_path = temp.path().join("exports/app.txt");
let download = artifact_download_report(ArtifactDownloadArgs {
scope: scope.clone(),
artifact: "app.txt".to_owned(),
@ -6402,7 +6591,7 @@ mod tests {
let export = artifact_export_report(ArtifactExportArgs {
scope,
artifact: "app.txt".to_owned(),
to: PathBuf::from("/tmp/app.txt"),
to: export_path.clone(),
receiver_node: "node-b".to_owned(),
})
.unwrap();
@ -6438,8 +6627,25 @@ mod tests {
assert_eq!(export["export_plan"]["source_node"], "node-a");
assert_eq!(export["export_plan"]["receiver_node"], "node-b");
assert_eq!(export["export_plan"]["artifact"], "app.txt");
assert_eq!(export["export_plan"]["local_path"], "/tmp/app.txt");
assert_eq!(export["export_plan"]["local_bytes_written_by_cli"], false);
assert_eq!(
export["export_plan"]["local_path"].as_str().unwrap(),
export_path.to_string_lossy().as_ref()
);
assert_eq!(export["export_plan"]["artifact_size_bytes"], 9);
assert_eq!(export["export_plan"]["local_bytes_written_by_cli"], true);
assert_eq!(
export["export_plan"]["local_export_status"],
"local_bytes_written"
);
assert_eq!(export["export_plan"]["bytes_written"], 9);
assert_eq!(
export["local_export"]["stream"]["content_material_returned_in_report"],
false
);
assert_eq!(
std::fs::read(&export_path).unwrap().as_slice(),
b"app-bytes"
);
assert_eq!(
export["export_plan"]["default_durable_store_assumed"],
false

1
crates/disasmer-coordinator/Cargo.toml
View file

@ -6,6 +6,7 @@ license.workspace = true
repository.workspace = true
[dependencies]
base64.workspace = true
disasmer-core = { path = "../disasmer-core" }
postgres.workspace = true
serde.workspace = true

53
crates/disasmer-coordinator/src/service.rs
View file

@ -3,6 +3,7 @@ use std::io::{BufRead, BufReader, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _};
use disasmer_core::{
Actor, AgentId, ArtifactId, ArtifactRegistry, Authorization, Capability, CapabilityReportError,
CredentialKind, DataPlaneObject, DataPlaneScope, DefaultScheduler, Digest,
@ -673,11 +674,17 @@ pub enum CoordinatorResponse {
link: DownloadLink,
streamed_bytes: u64,
charged_download_bytes: u64,
content_bytes_available: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
content_base64: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
content_source: Option<String>,
},
ArtifactExportPlan {
plan: DirectBulkTransferPlan,
source_node: NodeId,
receiver_node: NodeId,
artifact_size_bytes: u64,
},
Error {
message: String,
@ -2163,10 +2170,16 @@ impl CoordinatorService {
} => {
let _ = token_nonce;
let context = user_context(tenant, project, actor_user);
let artifact = ArtifactId::new(artifact);
let policy = DownloadPolicy { max_bytes };
let downloadable_size = self
.artifact_registry
.downloadable_size(&context, &artifact, &policy)?;
let chunk_bytes = chunk_bytes.min(downloadable_size);
let mut stream = self.artifact_registry.open_download_stream(
&context,
&ArtifactId::new(artifact),
&DownloadPolicy { max_bytes },
&artifact,
&policy,
&token_digest,
now_epoch_seconds,
&self.download_limits,
@ -2181,10 +2194,18 @@ impl CoordinatorService {
)?;
let charged_download_bytes =
self.download_meter.used(&LimitKind::ArtifactDownloadBytes);
let content_base64 =
self.download_stream_content_base64(&context, &artifact, stream.streamed_bytes);
let content_bytes_available = content_base64.is_some();
let content_source =
content_bytes_available.then(|| "coordinator_complete_stdout_tail".to_owned());
Ok(CoordinatorResponse::ArtifactDownloadStream {
link: stream.link,
streamed_bytes: stream.streamed_bytes,
charged_download_bytes,
content_bytes_available,
content_base64,
content_source,
})
}
CoordinatorRequest::RevokeArtifactDownloadLink {
@ -2254,11 +2275,39 @@ impl CoordinatorService {
plan,
source_node,
receiver_node,
artifact_size_bytes: metadata.size,
})
}
}
}
fn download_stream_content_base64(
&self,
context: &disasmer_core::AuthContext,
artifact: &ArtifactId,
streamed_bytes: u64,
) -> Option<String> {
self.task_events
.iter()
.rev()
.find(|event| {
event.tenant == context.tenant
&& event.project == context.project
&& event.status_code == Some(0)
&& !event.stdout_truncated
&& event.stdout_bytes == event.stdout_tail.as_bytes().len() as u64
&& event.artifact_size_bytes == Some(event.stdout_bytes)
&& event.artifact_size_bytes == Some(streamed_bytes)
&& event
.artifact_path
.as_ref()
.map(artifact_id_from_path)
.as_ref()
== Some(artifact)
})
.map(|event| BASE64_STANDARD.encode(event.stdout_tail.as_bytes()))
}
fn record_debug_audit_event(
&mut self,
tenant: TenantId,