vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

hive-c0re: finish chrono-clock migration

Browse source
This commit is contained in:
damocles 2026-08-01 23:55:00 +02:00 • committed by mara
commit 9293c5d580
14 changed files with 82 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

14
hive-c0re/src/stores/approvals.rs
View file

@ -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)
}

7
hive-c0re/src/stores/audit_log.rs
View file

@ -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<AuditEntry> {
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<u64> {
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();
}

16
hive-c0re/src/stores/broker.rs
View file

@ -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<u64> {
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<i64> = rows.iter().map(|(id, _, _, _, _)| *id).collect();
let placeholders = std::iter::repeat_n("?", ids.len())
.collect::<Vec<_>>()
@ -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<u64> {
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

10
hive-c0re/src/stores/build_logs.rs
View file

@ -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<i64> {
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<u64> {
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(

7
hive-c0re/src/stores/operator_questions.rs
View file

@ -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))
}

4
hive-c0re/src/stores/power.rs
View file

@ -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(())

56
hive-c0re/src/stores/scheduled_prompts.rs
View file

@ -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<u64>,
pub next_fire_at_unix: i64,
pub created_at_unix: i64,
pub next_fire_at_unix: DateTime<Utc>,
pub created_at_unix: DateTime<Utc>,
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<i64>,
pub cancelled_at_unix: Option<DateTime<Utc>>,
pub description: Option<String>,
/// Set while the schedule is paused. Worker skips rows where
/// `paused_at_unix IS NOT NULL`. Cleared by `resume()`.
pub paused_at_unix: Option<i64>,
pub paused_at_unix: Option<DateTime<Utc>>,
pub targets: Vec<ScheduleTarget>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScheduleTarget {
pub target: String,
pub cancelled_at_unix: Option<i64>,
pub last_fired_at_unix: Option<i64>,
pub cancelled_at_unix: Option<DateTime<Utc>>,
pub last_fired_at_unix: Option<DateTime<Utc>>,
pub last_result: Option<String>,
}
@ -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<Schedule> {
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<i64>>(7)?
.map(hive_sh4re::wire_time::from_secs),
description: row.get(8)?,
paused_at_unix: row.get(9)?,
paused_at_unix: row
.get::<_, Option<i64>>(9)?
.map(hive_sh4re::wire_time::from_secs),
targets: Vec::new(),
})
}
@ -673,8 +677,12 @@ fn load_targets(conn: &Connection, schedule_id: i64) -> Result<Vec<ScheduleTarge
let rows = stmt.query_map(params![schedule_id], |row| {
Ok(ScheduleTarget {
target: row.get(0)?,
cancelled_at_unix: row.get(1)?,
last_fired_at_unix: row.get(2)?,
cancelled_at_unix: row
.get::<_, Option<i64>>(1)?
.map(hive_sh4re::wire_time::from_secs),
last_fired_at_unix: row
.get::<_, Option<i64>>(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]