feat(scheduler): support durable schedule control plane

proto/trogonai/scheduler/schedules/projections/v1/schedule.proto

@@ -0,0 +1,38 @@
+edition = "2024";
+
+package trogonai.scheduler.schedules.projections.v1;
+
+import "google/protobuf/timestamp.proto";
+import "trogonai/scheduler/schedules/v1/delivery.proto";
+import "trogonai/scheduler/schedules/v1/message.proto";
+import "trogonai/scheduler/schedules/v1/schedule.proto";
+import "trogonai/scheduler/schedules/v1/schedule_status.proto";
+
+// Schedule is the read-model projection of a schedule's current state, folded
+// from the schedule event stream and stored as the value of each entry in the
+// schedules KV bucket.
+//
+// It is intentionally independent of schedules.v1.Schedule (the scheduling
+// strategy, which appears here only as a nested field): this is a derived,
+// rebuildable read view, so its shape evolves on its own under protobuf's field
+// rules. Keeping it in its own package isolates those changes from the event and
+// command schemas.
+message Schedule {
+  // Stable schedule id; also the schedule stream id.
+  string schedule_id = 1 [features.field_presence = LEGACY_REQUIRED];
+  // Current lifecycle status (scheduled or paused).
+  trogonai.scheduler.schedules.v1.ScheduleStatus status = 2 [features.field_presence = LEGACY_REQUIRED];
+  // True once a recurring schedule has run to exhaustion: it stays visible but
+  // will never fire again.
+  bool completed = 3;
+  // The next planned occurrence, if one is armed and pending.
+  google.protobuf.Timestamp next_occurrence_at = 4;
+  // The most recently recorded occurrence, if any has fired.
+  google.protobuf.Timestamp last_occurrence_at = 5;
+  // The schedule definition recorded at creation.
+  trogonai.scheduler.schedules.v1.Schedule schedule = 6 [features.field_presence = LEGACY_REQUIRED];
+  // The delivery definition recorded at creation.
+  trogonai.scheduler.schedules.v1.Delivery delivery = 7 [features.field_presence = LEGACY_REQUIRED];
+  // The static message payload and headers recorded at creation.
+  trogonai.scheduler.schedules.v1.Message message = 8 [features.field_presence = LEGACY_REQUIRED];
+}

rsworkspace/Cargo.toml

@@ -23,9 +23,12 @@ mcp-nats = { path = "crates/mcp-nats" }
 mcp-nats-server = { path = "crates/mcp-nats-server" }
 mcp-nats-stdio = { path = "crates/mcp-nats-stdio" }
 trogon-nats = { path = "crates/trogon-nats" }
+trogon-decider = { path = "crates/trogon-decider" }
+trogon-decider-nats = { path = "crates/trogon-decider-nats" }
 trogon-decider-runtime = { path = "crates/trogon-decider-runtime" }
 trogon-service-config = { path = "crates/trogon-service-config" }
 trogon-std = { path = "crates/trogon-std" }
+trogonai-proto = { path = "crates/trogonai-proto" }
 
 # A2A
 a2a = { package = "a2a-lf", version = "=0.3.0" }
@@ -85,8 +88,10 @@ tonic = { version = "=0.14.6", default-features = false, features = ["transport"
 moka = { version = "=0.12.10", features = ["future"] }
 
 # Misc
+chrono-tz = "=0.10.4"
 http = "=1.4.0"
 reqwest = { version = "=0.12.28", default-features = false, features = ["json", "rustls-tls", "stream"] }
+rrule = "=0.14.0"
 sha2 = "=0.10.8"
 wiremock = "=0.6.3"
 tempfile = "=3.27.0"

rsworkspace/crates/trogon-scheduler/Cargo.toml

@@ -4,32 +4,45 @@ version = "0.1.0"
 edition = "2024"
 license = "Apache-2.0"
 publish = false
+autotests = false
 
-[lints]
-workspace = true
+[lints.rust]
+warnings = "deny"
+unexpected_cfgs = { level = "deny", check-cfg = ['cfg(coverage)'] }
+
+[lints.clippy]
+all = "deny"
+expect_used = "deny"
+panic = "deny"
+unwrap_used = "deny"
+
+[features]
+test-support = ["trogon-decider-runtime/test-support", "trogon-nats/test-support"]
 
 [dependencies]
 thiserror = { workspace = true }
+trogon-telemetry = { workspace = true }
 async-nats = { workspace = true, features = ["jetstream", "kv", "server_2_12"] }
 base64 = { workspace = true }
 buffa = { workspace = true }
 buffa-types = { workspace = true }
 bytes = { workspace = true }
-chrono = { version = "0.4" }
-chrono-tz = "0.10"
+chrono = { version = "0.4", features = ["serde"] }
+chrono-tz = { workspace = true }
 cron = "0.15"
 futures = { workspace = true }
 opentelemetry = { workspace = true }
-rrule = "0.14"
+rrule = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
-tokio = { workspace = true, features = ["rt", "macros", "sync", "time"] }
+tokio = { workspace = true, features = ["time", "signal", "rt-multi-thread", "macros", "sync"] }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
-trogon-decider-runtime = { workspace = true }
 trogon-nats = { workspace = true }
-trogon-std = { workspace = true }
-trogonai-proto = { path = "../trogonai-proto", features = ["schedules"] }
+trogon-decider-nats = { workspace = true }
+trogon-decider-runtime = { workspace = true }
+trogonai-proto = { workspace = true, features = ["schedules"] }
+trogon-std = { workspace = true, features = ["uuid"] }
 uuid = { workspace = true, features = ["v5"] }
 
 [dev-dependencies]
@@ -38,7 +51,17 @@ proptest = "1"
 testcontainers-modules = { version = "0.15", features = ["nats"] }
 time = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] }
-trogon-decider = { path = "../trogon-decider", features = ["test-support"] }
-trogon-decider-nats = { path = "../trogon-decider-nats" }
+trogon-decider = { workspace = true, features = ["test-support"] }
+trogon-decider-runtime = { workspace = true, features = ["test-support"] }
+trogon-decider-nats = { workspace = true }
 trogon-nats = { workspace = true, features = ["test-support"] }
 trogon-std = { workspace = true, features = ["test-support"] }
+
+[[test]]
+name = "schedule_unit"
+path = "tests/schedule_unit.rs"
+required-features = ["test-support"]
+
+[[test]]
+name = "integration"
+path = "tests/integration.rs"

rsworkspace/crates/trogon-scheduler/src/config.rs

@@ -0,0 +1,102 @@
+use serde::{Deserialize, Serialize};
+use trogon_decider_runtime::{StreamPosition, StreamWritePrecondition};
+
+use crate::error::SchedulerError;
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "mode", content = "position", rename_all = "snake_case")]
+pub enum ScheduleWriteCondition {
+    MustNotExist,
+    MustBeAtPosition(StreamPosition),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ScheduleWriteState {
+    current_position: Option<StreamPosition>,
+    exists: bool,
+}
+
+impl ScheduleWriteState {
+    pub const fn new(current_position: Option<StreamPosition>, exists: bool) -> Self {
+        Self {
+            current_position,
+            exists,
+        }
+    }
+
+    pub const fn current_position(self) -> Option<StreamPosition> {
+        self.current_position
+    }
+
+    pub const fn exists(self) -> bool {
+        self.exists
+    }
+}
+
+impl ScheduleWriteCondition {
+    pub fn ensure(self, id: &str, state: ScheduleWriteState) -> Result<(), SchedulerError> {
+        match self {
+            Self::MustNotExist if !state.exists() => Ok(()),
+            Self::MustBeAtPosition(expected) if state.current_position() == Some(expected) => Ok(()),
+            expected => Err(SchedulerError::OptimisticConcurrencyConflict {
+                id: id.to_string(),
+                expected: expected.into(),
+                current_position: state.current_position(),
+            }),
+        }
+    }
+}
+
+impl From<ScheduleWriteCondition> for StreamWritePrecondition {
+    fn from(value: ScheduleWriteCondition) -> Self {
+        match value {
+            ScheduleWriteCondition::MustNotExist => Self::NoStream,
+            ScheduleWriteCondition::MustBeAtPosition(position) => Self::At(position),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn position(value: u64) -> StreamPosition {
+        StreamPosition::try_new(value).expect("test stream position must be non-zero")
+    }
+
+    #[test]
+    fn write_condition_ensures_expected_positions() {
+        ScheduleWriteCondition::MustNotExist
+            .ensure("alpha", ScheduleWriteState::new(None, false))
+            .unwrap();
+        ScheduleWriteCondition::MustBeAtPosition(position(3))
+            .ensure("alpha", ScheduleWriteState::new(Some(position(3)), true))
+            .unwrap();
+
+        let error = ScheduleWriteCondition::MustBeAtPosition(position(2))
+            .ensure("alpha", ScheduleWriteState::new(Some(position(4)), true))
+            .unwrap_err();
+        assert!(matches!(
+            error,
+            SchedulerError::OptimisticConcurrencyConflict {
+                current_position: Some(_),
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn write_condition_rejects_reusing_deleted_stream_ids() {
+        let error = ScheduleWriteCondition::MustNotExist
+            .ensure("alpha", ScheduleWriteState::new(Some(position(7)), true))
+            .unwrap_err();
+
+        assert!(matches!(
+            error,
+            SchedulerError::OptimisticConcurrencyConflict {
+                current_position: Some(_),
+                ..
+            }
+        ));
+    }
+}