posemesh_compute_node_runner_api/
runner.rs

1use crate::types::LeaseEnvelope;
2use anyhow::Result;
3use async_trait::async_trait;
4use std::path::PathBuf;
5
6/// Result of materializing a CID, including discovered metadata from the domain server.
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct MaterializedInput {
9    pub cid: String,
10    pub path: PathBuf,
11    pub data_id: Option<String>,
12    pub name: Option<String>,
13    pub data_type: Option<String>,
14    pub domain_id: Option<String>,
15    pub root_dir: PathBuf,
16    pub related_files: Vec<PathBuf>,
17    pub extracted_paths: Vec<PathBuf>,
18}
19
20impl MaterializedInput {
21    pub fn new(cid: impl Into<String>, path: PathBuf) -> Self {
22        let root_dir = path
23            .parent()
24            .map(|p| p.to_path_buf())
25            .unwrap_or_else(|| path.clone());
26        Self {
27            cid: cid.into(),
28            path,
29            data_id: None,
30            name: None,
31            data_type: None,
32            domain_id: None,
33            root_dir,
34            related_files: Vec::new(),
35            extracted_paths: Vec::new(),
36        }
37    }
38}
39
40/// Source of input artifacts for a task.
41#[async_trait]
42pub trait InputSource: Send + Sync {
43    /// Fetch object bytes by CID.
44    async fn get_bytes_by_cid(&self, cid: &str) -> Result<Vec<u8>>;
45
46    /// Materialize CID to a temporary file path and return its location.
47    async fn materialize_cid_to_temp(&self, cid: &str) -> Result<std::path::PathBuf>;
48
49    /// Materialize CID and include optional metadata describing the source artifact.
50    async fn materialize_cid_with_meta(&self, cid: &str) -> Result<MaterializedInput> {
51        let path = self.materialize_cid_to_temp(cid).await?;
52        Ok(MaterializedInput::new(cid, path))
53    }
54}
55
56/// Destination for task output artifacts.
57#[async_trait]
58pub trait ArtifactSink: Send + Sync {
59    /// Upload raw bytes under a relative path within the outputs prefix.
60    async fn put_bytes(&self, rel_path: &str, bytes: &[u8]) -> Result<()>;
61
62    /// Upload a file from the local filesystem under a relative path.
63    async fn put_file(&self, rel_path: &str, file_path: &std::path::Path) -> Result<()>;
64
65    /// Optional: open a multipart upload writer for a large artifact.
66    async fn open_multipart(&self, _rel_path: &str) -> Result<Box<dyn MultipartUpload>> {
67        Err(anyhow::anyhow!("multipart not supported"))
68    }
69}
70
71/// Handle returned by `ArtifactSink::open_multipart`.
72#[async_trait]
73pub trait MultipartUpload: Send + Sync {
74    /// Write a chunk of the artifact.
75    async fn write_chunk(&mut self, chunk: &[u8]) -> Result<()>;
76    /// Finish and commit the artifact.
77    async fn finish(self: Box<Self>) -> Result<()>;
78}
79
80/// Control-plane interface for a running task.
81#[async_trait]
82pub trait ControlPlane: Send + Sync {
83    /// Returns true if the task has been cancelled.
84    async fn is_cancelled(&self) -> bool;
85
86    /// Report progress to the engine; opaque JSON accepted to avoid coupling.
87    async fn progress(&self, value: serde_json::Value) -> Result<()>;
88
89    /// Log an event with fields to be attached to heartbeats.
90    async fn log_event(&self, fields: serde_json::Value) -> Result<()>;
91}
92
93/// Task context passed to runners.
94pub struct TaskCtx<'a> {
95    pub lease: &'a LeaseEnvelope,
96    pub input: &'a dyn InputSource,
97    pub output: &'a dyn ArtifactSink,
98    pub ctrl: &'a dyn ControlPlane,
99}
100
101/// Runner entrypoint.
102#[async_trait]
103pub trait Runner: Send + Sync {
104    /// Capability string this runner implements (e.g., "/reconstruction/legacy/v1").
105    fn capability(&self) -> &'static str;
106
107    /// Execute the task.
108    async fn run(&self, ctx: TaskCtx<'_>) -> Result<()>;
109}
posemesh_compute_node_runner_api/runner.rs

posemesh_compute_node_runner_api/
runner.rs
