progress: Emit a Start message with API version, drop from fields

cgwalters · cgwalters · commit 63d1199e4458 · 2025-01-08T15:23:04.000-05:00
And use a semantic version for the API version as this allows
a clearer evolution.

Signed-off-by: Colin Walters &lt;walters@verbum.org&gt;
diff --git a/docs/src/experimental-progress-fd.md b/docs/src/experimental-progress-fd.md
@@ -12,9 +12,8 @@ The format of data output over `--progress-fd` is [JSON Lines](https://jsonlines
 which is a series of JSON objects separated by newlines (the intermediate
 JSON content is guaranteed not to contain a literal newline).
 
-The current API version is `org.containers.bootc/progress/v1`. You can find
-the JSON schema describing this version here:
-[progress-v1.schema.json](progress-v1.schema.json).
+You can find the JSON schema describing this version here:
+[progress-v0.schema.json](progress-v0.schema.json).
 
 Deploying a new image with either switch or upgrade consists
 of three stages: `pulling`, `importing`, and `staging`. The `pulling` step
diff --git a/docs/src/progress-v0.schema.json b/docs/src/progress-v0.schema.json
@@ -3,11 +3,29 @@
   "title": "Event",
   "description": "An event emitted as JSON.",
   "oneOf": [
+    {
+      "type": "object",
+      "required": [
+        "type",
+        "version"
+      ],
+      "properties": {
+        "type": {
+          "type": "string",
+          "enum": [
+            "Start"
+          ]
+        },
+        "version": {
+          "description": "The semantic version of the progress protocol.",
+          "type": "string"
+        }
+      }
+    },
     {
       "description": "An incremental update to a container image layer download",
       "type": "object",
       "required": [
-        "api_version",
         "bytes",
         "bytes_cached",
         "bytes_total",
@@ -21,10 +39,6 @@
         "type"
       ],
       "properties": {
-        "api_version": {
-          "description": "The version of the progress event format.",
-          "type": "string"
-        },
         "bytes": {
           "description": "The number of bytes already fetched.",
           "type": "integer",
@@ -92,7 +106,6 @@
       "description": "An incremental update with discrete steps",
       "type": "object",
       "required": [
-        "api_version",
         "description",
         "id",
         "steps",
@@ -103,10 +116,6 @@
         "type"
       ],
       "properties": {
-        "api_version": {
-          "description": "The version of the progress event format.",
-          "type": "string"
-        },
         "description": {
           "description": "A human readable description of the task if i18n is not available.",
           "type": "string"
diff --git a/lib/src/deploy.rs b/lib/src/deploy.rs
@@ -21,7 +21,7 @@ use ostree_ext::ostree::{self, Sysroot};
 use ostree_ext::sysroot::SysrootLock;
 use ostree_ext::tokio_util::spawn_blocking_cancellable_flatten;
 
-use crate::progress_jsonl::{Event, ProgressWriter, SubTaskBytes, SubTaskStep, API_VERSION};
+use crate::progress_jsonl::{Event, ProgressWriter, SubTaskBytes, SubTaskStep};
 use crate::spec::ImageReference;
 use crate::spec::{BootOrder, HostSpec};
 use crate::status::labels_of_config;
@@ -214,7 +214,6 @@ async fn handle_layer_progress_print(
                         subtask.bytes = layer_size;
                         subtasks.push(subtask.clone());
                         prog.send(Event::ProgressBytes {
-                            api_version: API_VERSION.into(),
                             task: "pulling".into(),
                             description: format!("Pulling Image: {digest}").into(),
                             id: (*digest).into(),
@@ -245,7 +244,6 @@ async fn handle_layer_progress_print(
                     byte_bar.set_position(bytes.fetched);
                     subtask.bytes = byte_bar.position();
                     prog.send_lossy(Event::ProgressBytes {
-                        api_version: API_VERSION.into(),
                         task: "pulling".into(),
                         description: format!("Pulling Image: {digest}").into(),
                         id: (*digest).into(),
@@ -283,7 +281,6 @@ async fn handle_layer_progress_print(
     // use as a heuristic to begin import progress
     // Cannot be lossy or it is dropped
     prog.send(Event::ProgressSteps {
-        api_version: API_VERSION.into(),
         task: "importing".into(),
         description: "Importing Image".into(),
         id: (*digest).into(),
@@ -358,7 +355,6 @@ pub(crate) async fn pull(
     let prog = printer.await?;
     // Both the progress and the import are done, so import is done as well
     prog.send(Event::ProgressSteps {
-        api_version: API_VERSION.into(),
         task: "importing".into(),
         description: "Importing Image".into(),
         id: digest_imp.clone().as_ref().into(),
@@ -573,7 +569,6 @@ pub(crate) async fn stage(
     };
     let mut subtasks = vec![];
     prog.send(Event::ProgressSteps {
-        api_version: API_VERSION.into(),
         task: "staging".into(),
         description: "Deploying Image".into(),
         id: image.manifest_digest.clone().as_ref().into(),
@@ -596,7 +591,6 @@ pub(crate) async fn stage(
     subtask.description = "Deploying Image".into();
     subtask.completed = false;
     prog.send(Event::ProgressSteps {
-        api_version: API_VERSION.into(),
         task: "staging".into(),
         description: "Deploying Image".into(),
         id: image.manifest_digest.clone().as_ref().into(),
@@ -627,7 +621,6 @@ pub(crate) async fn stage(
     subtask.description = "Pulling Bound Images".into();
     subtask.completed = false;
     prog.send(Event::ProgressSteps {
-        api_version: API_VERSION.into(),
         task: "staging".into(),
         description: "Deploying Image".into(),
         id: image.manifest_digest.clone().as_ref().into(),
@@ -650,7 +643,6 @@ pub(crate) async fn stage(
     subtask.description = "Removing old images".into();
     subtask.completed = false;
     prog.send(Event::ProgressSteps {
-        api_version: API_VERSION.into(),
         task: "staging".into(),
         description: "Deploying Image".into(),
         id: image.manifest_digest.clone().as_ref().into(),
@@ -674,7 +666,6 @@ pub(crate) async fn stage(
     subtask.completed = true;
     subtasks.push(subtask.clone());
     prog.send(Event::ProgressSteps {
-        api_version: API_VERSION.into(),
         task: "staging".into(),
         description: "Deploying Image".into(),
         id: image.manifest_digest.clone().as_ref().into(),
diff --git a/lib/src/progress_jsonl.rs b/lib/src/progress_jsonl.rs
@@ -16,12 +16,15 @@ use tokio::sync::Mutex;
 // Maximum number of times per second that an event will be written.
 const REFRESH_HZ: u16 = 5;
 
-pub const API_VERSION: &str = "org.containers.bootc.progress/v1";
+/// Semantic version of the protocol.
+const API_VERSION: &str = "0.1.0";
 
 /// An incremental update to e.g. a container image layer download.
 /// The first time a given "subtask" name is seen, a new progress bar should be created.
 /// If bytes == bytes_total, then the subtask is considered complete.
-#[derive(Debug, serde::Serialize, serde::Deserialize, Default, Clone, JsonSchema)]
+#[derive(
+    Debug, serde::Serialize, serde::Deserialize, Default, Clone, JsonSchema, PartialEq, Eq,
+)]
 #[serde(rename_all = "camelCase")]
 pub struct SubTaskBytes<'t> {
     /// A machine readable type for the task (used for i18n).
@@ -45,7 +48,9 @@ pub struct SubTaskBytes<'t> {
 }
 
 /// Marks the beginning and end of a dictrete step
-#[derive(Debug, serde::Serialize, serde::Deserialize, Default, Clone, JsonSchema)]
+#[derive(
+    Debug, serde::Serialize, serde::Deserialize, Default, Clone, JsonSchema, PartialEq, Eq,
+)]
 #[serde(rename_all = "camelCase")]
 pub struct SubTaskStep<'t> {
     /// A machine readable type for the task (used for i18n).
@@ -65,18 +70,20 @@ pub struct SubTaskStep<'t> {
 }
 
 /// An event emitted as JSON.
-#[derive(Debug, serde::Serialize, serde::Deserialize, JsonSchema)]
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, JsonSchema, PartialEq, Eq)]
 #[serde(
     tag = "type",
     rename_all = "PascalCase",
     rename_all_fields = "camelCase"
 )]
 pub enum Event<'t> {
+    Start {
+        /// The semantic version of the progress protocol.
+        #[serde(borrow)]
+        version: Cow<'t, str>,
+    },
     /// An incremental update to a container image layer download
     ProgressBytes {
-        /// The version of the progress event format.
-        #[serde(borrow)]
-        api_version: Cow<'t, str>,
         /// A machine readable type (e.g., pulling) for the task (used for i18n
         /// and UI customization).
         #[serde(borrow)]
@@ -106,9 +113,6 @@ pub enum Event<'t> {
     },
     /// An incremental update with discrete steps
     ProgressSteps {
-        /// The version of the progress event format.
-        #[serde(borrow)]
-        api_version: Cow<'t, str>,
         /// A machine readable type (e.g., pulling) for the task (used for i18n
         /// and UI customization).
         #[serde(borrow)]
@@ -150,6 +154,8 @@ impl FromStr for RawProgressFd {
 
 #[derive(Debug)]
 struct ProgressWriterInner {
+    /// true if we sent the initial Start message
+    sent_start: bool,
     last_write: Option<std::time::Instant>,
     fd: BufWriter<Sender>,
 }
@@ -171,6 +177,7 @@ impl TryFrom<OwnedFd> for ProgressWriter {
 impl From<Sender> for ProgressWriter {
     fn from(value: Sender) -> Self {
         let inner = ProgressWriterInner {
+            sent_start: false,
             last_write: None,
             fd: BufWriter::new(value),
         };
@@ -190,6 +197,18 @@ impl TryFrom<RawProgressFd> for ProgressWriter {
 }
 
 impl ProgressWriter {
+    /// Serialize the target value as a single line of JSON and write it.
+    async fn send_impl_inner<T: Serialize>(inner: &mut ProgressWriterInner, v: T) -> Result<()> {
+        // serde is guaranteed not to output newlines here
+        let buf = serde_json::to_vec(&v)?;
+        inner.fd.write_all(&buf).await?;
+        // We always end in a newline
+        inner.fd.write_all(b"\n").await?;
+        // And flush to ensure the remote side sees updates immediately
+        inner.fd.flush().await?;
+        Ok(())
+    }
+
     /// Serialize the target object to JSON as a single line
     pub(crate) async fn send_impl<T: Serialize>(&self, v: T, required: bool) -> Result<()> {
         let mut guard = self.inner.lock().await;
@@ -198,8 +217,17 @@ impl ProgressWriter {
             return Ok(());
         };
 
+        // If this is our first message, emit the Start message
+        if !inner.sent_start {
+            inner.sent_start = true;
+            let start = Event::Start {
+                version: API_VERSION.into(),
+            };
+            Self::send_impl_inner(inner, &start).await?;
+        }
+
         // For messages that can be dropped, if we already sent an update within this cycle, discard this one.
-        // TODO: Also consider querying the pipe buffer and also dropping if we can't do this write.
+        // TODO: Also consider querying the pipe buffer and also dropping if wqe can't do this write.
         let now = Instant::now();
         if !required {
             const REFRESH_MS: u32 = 1000 / REFRESH_HZ as u32;
@@ -210,22 +238,15 @@ impl ProgressWriter {
             }
         }
 
-        // SAFETY: Propagating panics from the mutex here is intentional
-        // serde is guaranteed not to output newlines here
-        let buf = serde_json::to_vec(&v)?;
-        inner.fd.write_all(&buf).await?;
-        // We always end in a newline
-        inner.fd.write_all(b"\n").await?;
-        // And flush to ensure the remote side sees updates immediately
-        inner.fd.flush().await?;
+        Self::send_impl_inner(inner, &v).await?;
         // Update the last write time
         inner.last_write = Some(now);
         Ok(())
     }
 
     /// Send an event.
-    pub(crate) async fn send<T: Serialize>(&self, v: T) {
-        if let Err(e) = self.send_impl(v, true).await {
+    pub(crate) async fn send(&self, event: Event<'_>) {
+        if let Err(e) = self.send_impl(event, true).await {
             eprintln!("Failed to write to jsonl: {}", e);
             // Stop writing to fd but let process continue
             // SAFETY: Propagating panics from the mutex here is intentional
@@ -234,8 +255,8 @@ impl ProgressWriter {
     }
 
     /// Send an event that can be dropped.
-    pub(crate) async fn send_lossy<T: Serialize>(&self, v: T) {
-        if let Err(e) = self.send_impl(v, false).await {
+    pub(crate) async fn send_lossy(&self, event: Event<'_>) {
+        if let Err(e) = self.send_impl(event, false).await {
             eprintln!("Failed to write to jsonl: {}", e);
             // Stop writing to fd but let process continue
             // SAFETY: Propagating panics from the mutex here is intentional
@@ -272,18 +293,30 @@ mod test {
     #[tokio::test]
     async fn test_jsonl() -> Result<()> {
         let testvalues = [
-            S {
-                s: "foo".into(),
-                v: 42,
+            Event::ProgressSteps {
+                task: "sometask".into(),
+                description: "somedesc".into(),
+                id: "someid".into(),
+                steps_cached: 0,
+                steps: 0,
+                steps_total: 3,
+                subtasks: Vec::new(),
             },
-            S {
-                // Test with an embedded newline to sanity check that serde doesn't write it literally
-                s: "foo\nbar".into(),
-                v: 0,
+            Event::ProgressBytes {
+                task: "sometask".into(),
+                description: "somedesc".into(),
+                id: "someid".into(),
+                bytes_cached: 0,
+                bytes: 11,
+                bytes_total: 42,
+                steps_cached: 0,
+                steps: 0,
+                steps_total: 3,
+                subtasks: Vec::new(),
             },
         ];
         let (send, recv) = tokio::net::unix::pipe::pipe()?;
-        let testvalues_sender = &testvalues;
+        let testvalues_sender = testvalues.iter().cloned();
         let sender = async move {
             let w = ProgressWriter::try_from(send)?;
             for value in testvalues_sender {
@@ -296,10 +329,18 @@ mod test {
             let tf = BufReader::new(recv);
             let mut expected = testvalues.iter();
             let mut lines = tf.lines();
+            let mut got_first = false;
             while let Some(line) = lines.next_line().await? {
-                let found: S = serde_json::from_str(&line)?;
-                let expected = expected.next().unwrap();
-                assert_eq!(&found, expected);
+                let found: Event = serde_json::from_str(&line)?;
+                let expected_value = if !got_first {
+                    got_first = true;
+                    &Event::Start {
+                        version: API_VERSION.into(),
+                    }
+                } else {
+                    expected.next().unwrap()
+                };
+                assert_eq!(&found, expected_value);
             }
             anyhow::Ok(())
         };
diff --git a/tests/booted/test-image-pushpull-upgrade.nu b/tests/booted/test-image-pushpull-upgrade.nu
@@ -75,10 +75,12 @@ RUN echo test content > /usr/share/blah.txt
 
 # This just does some basic verification of the progress JSON
 def sanity_check_switch_progress_json [data] {
-    let first = $data.0;
     let event_count = $data | length
-    assert equal $first.type ProgressBytes
-    let steps = $first.stepsTotal
+    # The first one should always be a start event
+    let first = $data.0;
+    assert equal $first.type Start
+    let second = $data.1
+    let steps = $second.stepsTotal
     mut i = 0
     for elt in $data {
         if $elt.type != "ProgressBytes" {
diff --git a/xtask/src/xtask.rs b/xtask/src/xtask.rs
@@ -149,7 +149,7 @@ fn update_generated(sh: &Shell) -> Result<()> {
     }
     for (of, target) in [
         ("host", "docs/src/host-v1.schema.json"),
-        ("progress", "docs/src/progress-v1.schema.json"),
+        ("progress", "docs/src/progress-v0.schema.json"),
     ] {
         let schema = cmd!(sh, "cargo run -q -- internals print-json-schema --of={of}").read()?;
         std::fs::write(target, &schema)?;

Original file line number	Diff line number	Diff line change
`@@ -149,7 +149,7 @@ fn update_generated(sh: &Shell) -> Result<()> {`
`149`	`149`	`}`
`150`	`150`	`for (of, target) in [`
`151`	`151`	`("host", "docs/src/host-v1.schema.json"),`
`152`		`- ("progress", "docs/src/progress-v1.schema.json"),`
	`152`	`+ ("progress", "docs/src/progress-v0.schema.json"),`
`153`	`153`	`] {`
`154`	`154`	`let schema = cmd!(sh, "cargo run -q -- internals print-json-schema --of={of}").read()?;`
`155`	`155`	`std::fs::write(target, &schema)?;`