feat: close the executor side-channel gaps (#590, #591, #593)

crates/taskito-core/BINDING_CONTRACT.md

@@ -116,12 +116,14 @@ bytes of the section above, unchanged. `MAX_HEADER_BYTES` (64 KiB) and
 
 | Frame | Direction | Payload |
 |---|---|---|
-| `hello` | executor → scheduler | `{executor_id, sdk, version, tasks[], slots, protocol_version}` |
-| `hello_ack` | scheduler → executor | `{scheduler_id, protocol_version}` |
+| `hello` | executor → scheduler | `{executor_id, sdk, version, tasks[], slots, protocol_version, token?}` |
+| `hello_ack` | scheduler → executor | `{scheduler_id, protocol_version, capabilities[]}` |
 | `heartbeat` | executor → scheduler | `{free_slots}` |
-| `job` | scheduler → executor | `{id, task_name, payload_len, retry_count, max_retries, queue, timeout_ms, namespace}` + blob |
+| `job` | scheduler → executor | `{id, task_name, payload_len, retry_count, max_retries, queue, timeout_ms, namespace, disabled_middleware[], metadata}` + blob |
 | `cancel` | scheduler → executor | `{job_id}` |
 | `shutdown` | scheduler → executor | — |
+| `progress` | executor → scheduler | `{job_id, progress}` |
+| `task_log` | executor → scheduler | `{job_id, task_name, level, message, extra_len}` + blob |
 | `success` | executor → scheduler | `{job_id, result_len, task_name, wall_time_ns}` + blob |
 | `failure` | executor → scheduler | `{job_id, error, retry_count, max_retries, task_name, wall_time_ns, should_retry, timed_out}` |
 | `cancelled` | executor → scheduler | `{job_id, task_name, wall_time_ns}` |
@@ -132,9 +134,27 @@ Rules:
   is never silently downgraded. The scheduler sends `hello_ack` even when it is
   rejecting, so both ends can log both versions.
 - `result_len: null` means the task returned nothing; `0` means it returned an
-  empty value. They are distinct.
+  empty value. They are distinct. `extra_len` follows the same rule.
 - `should_retry` is the executor's decision — only it can see the exception. The
   core never inspects one.
+- **Optional behaviour is negotiated, not versioned.** `capabilities` lists what
+  the scheduler will do on an executor's behalf; `side_channel` means it applies
+  `progress` and `task_log` frames to storage. An executor sends neither frame
+  unless it was advertised. Adding one never bumps `protocol_version`, which
+  would force scheduler and executors to upgrade together.
+- **An unknown frame type is skipped, not fatal.** A reader that cannot name a
+  frame reads its declared payload length, discards that many bytes, logs once
+  and continues — the stream stays aligned, and a session keeps its in-flight
+  jobs. A frame type whose name *is* known but whose header will not parse stays
+  an error: that is a desync, not a newer peer.
+- **A new frame type must declare its payload length as `payload_len`.** It is
+  the only field a peer that predates the frame can find. `result_len` and
+  `extra_len` predate this rule and are read as equivalents.
+- `progress` and `task_log` are fire-and-forget: they carry no reply, never
+  settle a job, and a scheduler drops one naming a job the sender is not
+  running. `progress` is 0–100; a value outside that range is never written —
+  SDK boundaries clamp where they can, and the scheduler drops what still
+  reaches it rather than failing the job over a progress report.
 
 ## Task errors (structured, cross-SDK)
 When a task raises, the shell reports the failure as a **canonical JSON object**

crates/taskito-core/Cargo.toml

@@ -40,6 +40,10 @@ redis = { workspace = true, optional = true }
 
 [dev-dependencies]
 tempfile = "3"
+# Hand-rolled wire frames in the executor-attach tests: a frame type this build
+# has no variant for is exactly what the forward-compatibility path is about, so
+# the test has to be able to serialize one.
+serde = { workspace = true }
 # Raw connection access in the Redis contract tests, to set up race-window
 # state and assert on internal index keys. Only used by `#[cfg(feature = "redis")]`
 # tests; harmless when the feature is off.

crates/taskito-core/src/worker/executor.rs

@@ -30,7 +30,7 @@
 //! # }
 //! ```
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, PoisonError};
 use std::thread::{self, JoinHandle};
@@ -41,8 +41,8 @@ use tokio::sync::mpsc::error::TrySendError;
 
 use super::auth::Secret;
 use super::protocol::{
-    ExecutorMessage, FrameReader, FrameWriter, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL,
-    PROTOCOL_VERSION,
+    ExecutorMessage, FrameReader, FrameWriter, Incoming, ProtocolError, SchedulerMessage,
+    CAP_SIDE_CHANNEL, PROTOCOL_VERSION,
 };
 use super::transport::{Connection, ReadHalf, Transport, WriteHalf};
 use super::WorkerDispatcher;
@@ -380,6 +380,17 @@ impl ExecutorSideChannel {
         self.shared.side_channel
     }
 
+    /// Task log lines shed since this executor attached, because they were
+    /// produced faster than they could be framed.
+    ///
+    /// Progress has no equivalent: it coalesces, so a backlog collapses instead
+    /// of losing anything. Exposed because "the log line I wrote is not in the
+    /// dashboard" is otherwise indistinguishable from a bug, and an SDK can
+    /// surface this as the backpressure signal it is.
+    pub fn dropped_task_logs(&self) -> u64 {
+        self.shared.dropped_logs.load(Ordering::Relaxed)
+    }
+
     /// Middleware the operator has disabled for a running job's task.
     ///
     /// Resolved by the scheduler at dispatch and carried on the job frame, so
@@ -806,19 +817,34 @@ fn spawn_reader(
     thread::Builder::new()
         .name("taskito-executor-reader".to_string())
         .spawn(move || {
+            // Once per type per session: a newer scheduler may send the frame we
+            // cannot read on every dispatch.
+            let mut reported_unknown: HashSet<String> = HashSet::new();
             loop {
-                match reader.read::<SchedulerMessage>() {
-                    Ok((SchedulerMessage::Shutdown, _)) => {
+                match reader.read_or_skip::<SchedulerMessage>() {
+                    Ok(Incoming::Known(SchedulerMessage::Shutdown, _)) => {
                         log::info!(
                             "[taskito] scheduler asked executor {} to shut down",
                             shared.executor_id
                         );
                         break;
                     }
-                    Ok((SchedulerMessage::Cancel { job_id }, _)) => {
+                    Ok(Incoming::Known(SchedulerMessage::Cancel { job_id }, _)) => {
                         dispatcher.notify_cancel(&job_id);
                     }
-                    Ok((frame, payload)) => accept_job(&shared, frame, payload),
+                    Ok(Incoming::Known(frame, payload)) => accept_job(&shared, frame, payload),
+                    // A frame from a scheduler newer than this executor. The
+                    // session is otherwise fine, and ending it would fail every
+                    // job in flight over a frame this build never needed.
+                    Ok(Incoming::Unknown { frame_type }) => {
+                        if reported_unknown.insert(frame_type.clone()) {
+                            log::warn!(
+                                "[taskito] scheduler sent a '{frame_type}' frame executor {} does \
+                                 not know; ignoring it (upgrade the executor to use it)",
+                                shared.executor_id
+                            );
+                        }
+                    }
                     Err(ProtocolError::Eof) => {
                         log::info!(
                             "[taskito] scheduler closed the connection to executor {}",

crates/taskito-core/src/worker/mod.rs

@@ -19,7 +19,8 @@ pub use executor::{
     ExecutorSideChannel,
 };
 pub use protocol::{
-    Dispatch, ExecutorMessage, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL, PROTOCOL_VERSION,
+    Dispatch, ExecutorMessage, Incoming, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL,
+    PROTOCOL_VERSION,
 };
 pub use registry::{TaskError, TaskHandler, TaskRegistry, TaskResult};
 pub use remote::{AttachError, AttachedExecutor, Capacity, RemoteConfig, RemoteDispatcher};