From 9293c5d580f9a244e52d7bc927b16c9c8c504fc3 Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 1 Aug 2026 23:55:00 +0200 Subject: [PATCH] hive-c0re: finish chrono-clock migration --- hive-c0re/src/dashboard/state_snapshot.rs | 2 +- hive-c0re/src/job_queue/mod.rs | 11 ++-- hive-c0re/src/job_queue/model.rs | 3 +- hive-c0re/src/loose_ends.rs | 6 +- hive-c0re/src/socket_server/schedules.rs | 12 ++-- hive-c0re/src/stats/hive_stats.rs | 6 +- hive-c0re/src/stores/approvals.rs | 14 ++--- hive-c0re/src/stores/audit_log.rs | 7 +-- hive-c0re/src/stores/broker.rs | 16 +++--- hive-c0re/src/stores/build_logs.rs | 10 ++-- hive-c0re/src/stores/operator_questions.rs | 7 +-- hive-c0re/src/stores/power.rs | 4 +- hive-c0re/src/stores/scheduled_prompts.rs | 56 +++++++++++-------- .../src/workers/scheduled_prompts_worker.rs | 6 +- 14 files changed, 82 insertions(+), 78 deletions(-) diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 04c43935..27309cc0 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -538,7 +538,7 @@ fn build_transient_views( // Clamped at 0: `since` is wall-clock now (the node's own // `started_at`), so a backwards clock adjustment could otherwise // render a negative age. - secs: (hive_sh4re::wire_time::from_secs(hive_sh4re::wire_time::now_unix()) - st.since) + secs: (chrono::Utc::now() - st.since) .num_seconds() .max(0) .cast_unsigned(), diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index ada7faad..bbe90925 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -44,7 +44,6 @@ use hive_host_sock::jobs::NodeView; use hive_jobq::resources::ResourceTable; use hive_jobq::scheduler::{Outcome, Scheduler}; use hive_jobq::{Dep, Graph, NodeId}; -use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; pub use hive_jobq::TerminalState; @@ -107,7 +106,7 @@ struct NodeRuntime { struct DagMeta { source: Source, reason: String, - created_at: i64, + created_at: DateTime, } /// The mutable queue state behind the mutex: the crate scheduler plus the @@ -230,7 +229,7 @@ impl JobQueue { NodeKind::Dag { source: spec.source, reason: spec.reason, - created_at: now_unix(), + created_at: Utc::now(), }, Vec::new(), None, @@ -469,9 +468,7 @@ impl JobQueue { // `started_at` is set when a node enters `Running`, and this // only sees `Running` nodes — the fallback is unreachable in // practice, and "just now" is the honest answer if it isn't. - since: n - .started_at - .unwrap_or_else(|| hive_sh4re::wire_time::from_secs(now_unix())), + since: n.started_at.unwrap_or_else(Utc::now), }) }) .collect() @@ -628,7 +625,7 @@ impl QueueInner { id: container.get(), source: meta.source, reason: meta.reason.clone(), - created_at: hive_sh4re::wire_time::from_secs(meta.created_at), + created_at: meta.created_at, started_at: started.into_iter().min(), finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(), nodes, diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 96fba3c3..77ba0757 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -12,6 +12,7 @@ //! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! full design. +use chrono::{DateTime, Utc}; pub use hive_host_sock::jobs::{DagView, NodeId, PermPayload, Source, State}; use serde::Serialize; @@ -283,7 +284,7 @@ pub enum NodeKind { Dag { source: Source, reason: String, - created_at: i64, + created_at: DateTime, }, } diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index 621419fb..b20ef567 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -20,10 +20,10 @@ //! (de)serialisation, not the read. use anyhow::Result; +use chrono::Utc; use hive_sh4re::LooseEnd; use crate::coordinator::Coordinator; -use hive_sh4re::wire_time::now_unix; /// Open threads pending against `agent`: /// - undelivered inbox messages this agent still owes itself a `recv` @@ -42,7 +42,7 @@ use hive_sh4re::wire_time::now_unix; /// Propagates errors from `count_pending` and the pending-approval /// sqlite query. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { - let now = now_unix(); + let now = Utc::now().timestamp(); let mut out = Vec::new(); // Undelivered inbox messages this agent still owes itself a `recv` // for. Surfaced first (most actionable) and only when non-zero so a @@ -81,7 +81,7 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { /// only; sub-agents can't see each other's threads via the agent surface /// (`for_agent` filters by name). pub fn hive_wide(coord: &Coordinator) -> Result> { - let now = now_unix(); + let now = Utc::now().timestamp(); let mut out = Vec::new(); for a in coord.approvals.pending()? { out.push(LooseEnd::Approval { diff --git a/hive-c0re/src/socket_server/schedules.rs b/hive-c0re/src/socket_server/schedules.rs index bebe855d..aa43e99c 100644 --- a/hive-c0re/src/socket_server/schedules.rs +++ b/hive-c0re/src/socket_server/schedules.rs @@ -321,8 +321,8 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSc owner: s.owner, body: s.body, interval_seconds: s.interval_seconds, - next_fire_at_unix: hive_sh4re::wire_time::from_secs(s.next_fire_at_unix), - created_at_unix: hive_sh4re::wire_time::from_secs(s.created_at_unix), + next_fire_at_unix: s.next_fire_at_unix, + created_at_unix: s.created_at_unix, source: match s.source { crate::scheduled_prompts::ScheduleSource::Operator => { hive_sh4re::WireScheduleSource::Operator @@ -331,16 +331,16 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSc hive_sh4re::WireScheduleSource::Approval { id } } }, - cancelled_at_unix: s.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), - paused_at_unix: s.paused_at_unix.map(hive_sh4re::wire_time::from_secs), + cancelled_at_unix: s.cancelled_at_unix, + paused_at_unix: s.paused_at_unix, description: s.description, targets: s .targets .into_iter() .map(|t| hive_sh4re::WireScheduleTarget { target: t.target, - cancelled_at_unix: t.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), - last_fired_at_unix: t.last_fired_at_unix.map(hive_sh4re::wire_time::from_secs), + cancelled_at_unix: t.cancelled_at_unix, + last_fired_at_unix: t.last_fired_at_unix, last_result: t.last_result, }) .collect(), diff --git a/hive-c0re/src/stats/hive_stats.rs b/hive-c0re/src/stats/hive_stats.rs index 7c4fea23..94b1e8c9 100644 --- a/hive-c0re/src/stats/hive_stats.rs +++ b/hive-c0re/src/stats/hive_stats.rs @@ -25,7 +25,7 @@ use serde::{Deserialize, Serialize}; use utoipa::ToSchema; use crate::coordinator::Coordinator; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; /// Window accepted by `/api/stats-hive?window=`. Maps to a lookback /// span; the hive view is a flat rollup (no per-bucket trend — the @@ -349,7 +349,7 @@ fn read_bash_heads(conn: &Connection, from: i64) -> HashMap { /// per-agent db is skipped (logged), never fatal. #[must_use] pub fn hive_snapshot(window: Window, prices: &PriceTable) -> HiveStats { - let now = now_unix(); + let now = Utc::now().timestamp(); // Fixed windows look back a constant span; `all` aggregates every // recorded turn (`from == 0`). The hive rollup isn't time-bucketed, so // unlike the per-agent snapshot it needs no adaptive bucket sizing. @@ -466,7 +466,7 @@ mod tests { let conn = Connection::open_in_memory().unwrap(); conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);") .unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); for (ts, head) in [ (now - 100, "cargo"), (now - 200, "cargo"), diff --git a/hive-c0re/src/stores/approvals.rs b/hive-c0re/src/stores/approvals.rs index 313f04e4..651af265 100644 --- a/hive-c0re/src/stores/approvals.rs +++ b/hive-c0re/src/stores/approvals.rs @@ -7,7 +7,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result, bail}; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus}; use rusqlite::{Connection, OptionalExtension, params}; @@ -95,7 +95,7 @@ impl Approvals { agent, kind.as_str(), commit_ref, - now_unix(), + Utc::now().timestamp(), description, submitter, fetched_sha, @@ -206,7 +206,7 @@ impl Approvals { if row.status != "pending" { bail!("approval {id} is {}, not pending", row.status); } - let resolved_at = now_unix(); + let resolved_at = Utc::now().timestamp(); conn.execute( "UPDATE approvals SET status = 'approved', resolved_at = ?1 WHERE id = ?2", params![resolved_at, id], @@ -230,7 +230,7 @@ impl Approvals { let affected = conn.execute( "UPDATE approvals SET status = 'denied', resolved_at = ?1, note = ?2 WHERE id = ?3 AND status = 'pending'", - params![now_unix(), note, id], + params![Utc::now().timestamp(), note, id], )?; if affected == 0 { bail!("approval {id} not pending"); @@ -242,7 +242,7 @@ impl Approvals { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2 WHERE id = ?3", - params![now_unix(), note, id], + params![Utc::now().timestamp(), note, id], )?; Ok(()) } @@ -267,7 +267,7 @@ impl Approvals { if row.status != "pending" { bail!("approval {id} is {}, not pending", row.status); } - let resolved_at = now_unix(); + let resolved_at = Utc::now().timestamp(); let note = format!("cancelled by {canceller}"); tx.execute( "UPDATE approvals SET status = 'cancelled', resolved_at = ?1, note = ?2 WHERE id = ?3", @@ -295,7 +295,7 @@ impl Approvals { let n = conn.execute( "UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2 WHERE agent = ?3 AND status = 'pending'", - params![now_unix(), note, agent], + params![Utc::now().timestamp(), note, agent], )?; Ok(n) } diff --git a/hive-c0re/src/stores/audit_log.rs b/hive-c0re/src/stores/audit_log.rs index 8e8bc160..eb1cd377 100644 --- a/hive-c0re/src/stores/audit_log.rs +++ b/hive-c0re/src/stores/audit_log.rs @@ -25,7 +25,6 @@ use std::sync::{Arc, Mutex, OnceLock}; use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; -use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, params}; use serde::Serialize; @@ -141,7 +140,7 @@ impl AuditLog { outcome: AuditOutcome, detail: Option<&str>, ) -> Option { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); match conn.execute( "INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail) @@ -208,7 +207,7 @@ impl AuditLog { /// # Errors /// Returns an error if the `DELETE` query fails. pub fn vacuum(&self) -> Result { - let cutoff = now_unix() - KEEP_SECS; + let cutoff = Utc::now().timestamp() - KEEP_SECS; let conn = self.conn.lock().unwrap(); let removed = conn.execute("DELETE FROM audit_log WHERE ts_unix < ?1", params![cutoff])?; Ok(u64::try_from(removed).unwrap_or(0)) @@ -311,7 +310,7 @@ mod tests { let conn = db.conn.lock().unwrap(); conn.execute( "UPDATE audit_log SET ts_unix = ?1", - params![now_unix() - KEEP_SECS - 60], + params![Utc::now().timestamp() - KEEP_SECS - 60], ) .unwrap(); } diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 729bbeed..76e7dd51 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -6,8 +6,8 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; +use chrono::Utc; -use hive_sh4re::wire_time::now_unix; use hive_sh4re::{InboxRow, Message}; use crate::db::Migration; @@ -211,7 +211,7 @@ impl Broker { pub fn send(&self, message: &Message) -> Result<()> { let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); // Operator messages get elevated priority so they surface before // queued wakes (bash completions, forge events, etc.) when the // harness pops the next turn driver. All other senders stay at 0. @@ -391,7 +391,7 @@ impl Broker { const PREFIX: &str = "your parent changed from "; const SEPARATOR: &str = " to "; let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let existing: Option<(i64, String)> = conn .query_row( "SELECT id, body FROM messages @@ -513,7 +513,7 @@ impl Broker { /// `requeue_inflight`, the latter because they're still in flight /// from the broker's POV. Returns the number of rows removed. pub fn vacuum_delivered(&self, older_than_secs: i64) -> Result { - let cutoff = now_unix() - older_than_secs; + let cutoff = Utc::now().timestamp() - older_than_secs; let conn = self.conn.lock().unwrap(); let n = conn.execute( "DELETE FROM messages @@ -578,7 +578,7 @@ impl Broker { } // Stamp all popped rows in a single UPDATE — under the broker // mutex, well within sqlite's 999-param default. - let now = now_unix(); + let now = Utc::now().timestamp(); let ids: Vec = rows.iter().map(|(id, _, _, _, _)| *id).collect(); let placeholders = std::iter::repeat_n("?", ids.len()) .collect::>() @@ -642,7 +642,7 @@ impl Broker { if ids.is_empty() { return Ok(0); } - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); // Bind every id explicitly. Caps in the hundreds in the worst // case (a single very chatty turn); well under sqlite's 999 @@ -688,7 +688,7 @@ impl Broker { let n = conn.execute( "UPDATE messages SET acked_at = ?1 WHERE recipient = ?2 AND id <= ?3 AND acked_at IS NULL", - params![now_unix(), recipient, up_to], + params![Utc::now().timestamp(), recipient, up_to], )?; Ok(u64::try_from(n).unwrap_or(0)) } @@ -762,7 +762,7 @@ impl Broker { pub fn mark_all_read(&self, recipient: &str) -> Result { let mut inflight = self.inflight.lock().unwrap(); let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); // Two-axis update in one statement: set acked_at on every // row for the recipient that doesn't have it yet, AND backfill // delivered_at if it was NULL so the row is fully consumed diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs index 678a479a..24b302bd 100644 --- a/hive-c0re/src/stores/build_logs.rs +++ b/hive-c0re/src/stores/build_logs.rs @@ -7,7 +7,7 @@ use std::path::Path; use std::sync::{Arc, Mutex, OnceLock}; use anyhow::{Context, Result}; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; use rusqlite::{Connection, OptionalExtension, params}; use serde::Serialize; use tokio::sync::broadcast; @@ -170,7 +170,7 @@ impl BuildLogs { /// — the caller threads it through `append_stdout` / `append_stderr` /// while the child runs and into `finish` once it exits. pub fn start(&self, agent: &str, kind: &str, cmdline: &str) -> Result { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO build_logs (agent, kind, cmdline, started_at) VALUES (?1, ?2, ?3, ?4)", @@ -220,7 +220,7 @@ impl BuildLogs { /// Finalize a build attempt. Sets `finished_at` to now and /// `status` to the terminal state. Best-effort. pub fn finish(&self, id: i64, status: BuildStatus) { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); if let Err(e) = conn.execute( "UPDATE build_logs SET finished_at = ?1, status = ?2 WHERE id = ?3", @@ -375,7 +375,7 @@ impl BuildLogs { /// a long-running build shouldn't disappear from its own log /// viewer mid-stream. pub fn vacuum(&self) -> Result { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); let fail_cutoff = now - KEEP_FAIL_SECS; let ok_cutoff = now - KEEP_OK_SECS; @@ -526,7 +526,7 @@ mod tests { // stays within KEEP_FAIL_SECS so it survives; old_fail goes // beyond; old_ok goes past KEEP_OK_SECS but inside // KEEP_FAIL_SECS — proves the per-status rule. - let now = now_unix(); + let now = Utc::now().timestamp(); { let conn = db.conn.lock().unwrap(); conn.execute( diff --git a/hive-c0re/src/stores/operator_questions.rs b/hive-c0re/src/stores/operator_questions.rs index 64f7b350..089eef90 100644 --- a/hive-c0re/src/stores/operator_questions.rs +++ b/hive-c0re/src/stores/operator_questions.rs @@ -15,7 +15,6 @@ use std::sync::Mutex; use anyhow::{Context, Result, bail}; use chrono::{DateTime, Utc}; -use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, OptionalExtension, params}; use serde::Serialize; @@ -119,7 +118,7 @@ impl OperatorQuestions { i64::from(multi), deadline_at, target, - now_unix(), + Utc::now().timestamp(), ], )?; Ok(conn.last_insert_rowid()) @@ -177,7 +176,7 @@ impl OperatorQuestions { } conn.execute( "UPDATE operator_questions SET answer = ?1, answered_at = ?2 WHERE id = ?3", - params![answer, now_unix(), id], + params![answer, Utc::now().timestamp(), id], )?; Ok((question, asker, target)) } @@ -225,7 +224,7 @@ impl OperatorQuestions { let sentinel = format!("[cancelled by {canceller}]"); conn.execute( "UPDATE operator_questions SET answer = ?1, answered_at = ?2 WHERE id = ?3", - params![sentinel, now_unix(), id], + params![sentinel, Utc::now().timestamp(), id], )?; Ok((question, asker, target)) } diff --git a/hive-c0re/src/stores/power.rs b/hive-c0re/src/stores/power.rs index a5900adb..6aa6e384 100644 --- a/hive-c0re/src/stores/power.rs +++ b/hive-c0re/src/stores/power.rs @@ -18,7 +18,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; use rusqlite::{Connection, OptionalExtension, params}; const SCHEMA: &str = " @@ -129,7 +129,7 @@ impl PowerStore { conn.execute( "INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, ?3) ON CONFLICT(agent) DO UPDATE SET wanted = ?2, updated_at = ?3", - params![agent, wanted.as_str(), now_unix()], + params![agent, wanted.as_str(), Utc::now().timestamp()], ) .context("upsert agent_power")?; Ok(()) diff --git a/hive-c0re/src/stores/scheduled_prompts.rs b/hive-c0re/src/stores/scheduled_prompts.rs index f0877a5c..75bff2a3 100644 --- a/hive-c0re/src/stores/scheduled_prompts.rs +++ b/hive-c0re/src/stores/scheduled_prompts.rs @@ -16,7 +16,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result, bail}; -use hive_sh4re::wire_time::now_unix; +use chrono::{DateTime, Utc}; use rusqlite::{Connection, OptionalExtension, params}; use serde::{Deserialize, Serialize}; @@ -78,25 +78,25 @@ pub struct Schedule { /// `None` = one-shot, deleted after first fire. /// `Some(n)` = recurring every `n` seconds. pub interval_seconds: Option, - pub next_fire_at_unix: i64, - pub created_at_unix: i64, + pub next_fire_at_unix: DateTime, + pub created_at_unix: DateTime, pub source: ScheduleSource, /// Set when the *entire* schedule was cancelled (all targets /// flipped, or operator cancel-all). Worker reaps these on the /// next pass. - pub cancelled_at_unix: Option, + pub cancelled_at_unix: Option>, pub description: Option, /// Set while the schedule is paused. Worker skips rows where /// `paused_at_unix IS NOT NULL`. Cleared by `resume()`. - pub paused_at_unix: Option, + pub paused_at_unix: Option>, pub targets: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ScheduleTarget { pub target: String, - pub cancelled_at_unix: Option, - pub last_fired_at_unix: Option, + pub cancelled_at_unix: Option>, + pub last_fired_at_unix: Option>, pub last_result: Option, } @@ -237,7 +237,7 @@ impl ScheduledPrompts { &new.body, new.interval_seconds.map(i64::try_from).and_then(Result::ok), new.first_fire_at_unix, - now_unix(), + Utc::now().timestamp(), new.source.to_db_string(), &new.description, ], @@ -472,7 +472,7 @@ impl ScheduledPrompts { // orphaned tombstone. Plus removing-then-adding is the // natural way to "restart history" on one target without // a separate flow. - let now = now_unix(); + let now = Utc::now().timestamp(); if let Some(remove) = patch.targets_remove.as_deref() { for target in remove { tx.execute( @@ -536,7 +536,7 @@ impl ScheduledPrompts { /// already-cancelled row. pub fn cancel_all(&self, id: i64) -> Result<()> { let mut conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let tx = conn.transaction()?; tx.execute( "UPDATE scheduled_prompts @@ -560,7 +560,7 @@ impl ScheduledPrompts { /// are silently skipped. pub fn cancel_targets(&self, id: i64, targets: &[String]) -> Result<()> { let mut conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let tx = conn.transaction()?; for target in targets { tx.execute( @@ -599,7 +599,7 @@ impl ScheduledPrompts { /// cancelled or does not exist (handlers downcast to emit 404). pub fn pause(&self, id: i64) -> Result<()> { let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let n = conn.execute( "UPDATE scheduled_prompts SET paused_at_unix = COALESCE(paused_at_unix, ?1) @@ -653,12 +653,16 @@ fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result { owner: row.get(1)?, body: row.get(2)?, interval_seconds: interval.and_then(|i| u64::try_from(i).ok()), - next_fire_at_unix: row.get(4)?, - created_at_unix: row.get(5)?, + next_fire_at_unix: hive_sh4re::wire_time::from_secs(row.get(4)?), + created_at_unix: hive_sh4re::wire_time::from_secs(row.get(5)?), source: ScheduleSource::from_db_string(&source_str), - cancelled_at_unix: row.get(7)?, + cancelled_at_unix: row + .get::<_, Option>(7)? + .map(hive_sh4re::wire_time::from_secs), description: row.get(8)?, - paused_at_unix: row.get(9)?, + paused_at_unix: row + .get::<_, Option>(9)? + .map(hive_sh4re::wire_time::from_secs), targets: Vec::new(), }) } @@ -673,8 +677,12 @@ fn load_targets(conn: &Connection, schedule_id: i64) -> Result>(1)? + .map(hive_sh4re::wire_time::from_secs), + last_fired_at_unix: row + .get::<_, Option>(2)? + .map(hive_sh4re::wire_time::from_secs), last_result: row.get(3)?, }) })?; @@ -771,7 +779,7 @@ mod tests { let skipped = db.rearm(id, 400).expect("rearm"); assert_eq!(skipped, 5); let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 460); + assert_eq!(s.next_fire_at_unix.timestamp(), 460); } #[test] @@ -793,7 +801,7 @@ mod tests { // the manual fire-now "reset timer" path uses now + interval. db.set_next_fire(id, 1_234).expect("set_next_fire"); let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 1_234); + assert_eq!(s.next_fire_at_unix.timestamp(), 1_234); } #[test] @@ -814,7 +822,7 @@ mod tests { let skipped = db.rearm(id, 100).expect("rearm"); assert_eq!(skipped, 0); let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 160); + assert_eq!(s.next_fire_at_unix.timestamp(), 160); } #[test] @@ -825,7 +833,7 @@ mod tests { assert_eq!(skipped, 0); // next_fire_at unchanged — one-shots are reaped via delete(). let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 100); + assert_eq!(s.next_fire_at_unix.timestamp(), 100); } #[test] @@ -878,7 +886,7 @@ mod tests { // Cancelled targets do NOT get last-result writes. assert!(alice.last_fired_at_unix.is_none()); assert!(alice.last_result.is_none()); - assert_eq!(bob.last_fired_at_unix, Some(200)); + assert_eq!(bob.last_fired_at_unix.map(|dt| dt.timestamp()), Some(200)); assert_eq!(bob.last_result.as_deref(), Some("ok")); } @@ -909,7 +917,7 @@ mod tests { assert_eq!(s.body, "new body"); assert_eq!(s.description.as_deref(), Some("old desc")); assert_eq!(s.interval_seconds, Some(60)); - assert_eq!(s.next_fire_at_unix, 100); + assert_eq!(s.next_fire_at_unix.timestamp(), 100); } #[test] diff --git a/hive-c0re/src/workers/scheduled_prompts_worker.rs b/hive-c0re/src/workers/scheduled_prompts_worker.rs index 4adb0743..b7dac081 100644 --- a/hive-c0re/src/workers/scheduled_prompts_worker.rs +++ b/hive-c0re/src/workers/scheduled_prompts_worker.rs @@ -6,11 +6,11 @@ use std::sync::Arc; use std::time::Duration; +use chrono::Utc; use hive_sh4re::Message; use crate::coordinator::Coordinator; use crate::scheduled_prompts::Schedule; -use hive_sh4re::wire_time::now_unix; /// Per-tick cap. Each schedule fires once per tick at most; /// 100/tick × 5s tick = sustained throughput cap of ~20/sec, @@ -47,7 +47,7 @@ pub fn spawn(coord: Arc) { } fn tick(coord: &Arc) { - let now = now_unix(); + let now = Utc::now().timestamp(); let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) { Ok(rows) => rows, Err(e) => { @@ -285,7 +285,7 @@ pub async fn fire_now( schedule_id: i64, reset_timer: bool, ) -> anyhow::Result { - let now = now_unix(); + let now = Utc::now().timestamp(); let schedule = coord .scheduled_prompts .get(schedule_id)?