refactor: single now_unix in hive_sh4re::wire_time
replaces 15 per-module copies (now_unix/now_secs) across hive-c0re and hive-ag3nt; wire_time already owns the epoch-seconds convention
This commit is contained in:
parent
d190420946
commit
c84028ddcf
16 changed files with 45 additions and 125 deletions
|
|
@ -16,6 +16,7 @@ use hive_claude::TokenUsage;
|
||||||
use rusqlite::{Connection, params};
|
use rusqlite::{Connection, params};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
const CHANNEL_CAPACITY: usize = 256;
|
const CHANNEL_CAPACITY: usize = 256;
|
||||||
/// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
|
/// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
|
||||||
|
|
@ -213,13 +214,6 @@ pub fn write_forge_cursor<S: std::hash::BuildHasher>(
|
||||||
write_harness_json(&v);
|
write_harness_json(&v);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
const SCHEMA: &str = "
|
const SCHEMA: &str = "
|
||||||
CREATE TABLE IF NOT EXISTS events (
|
CREATE TABLE IF NOT EXISTS events (
|
||||||
|
|
|
||||||
|
|
@ -7,6 +7,7 @@ use crate::events::Bus;
|
||||||
use crate::mcp::REDELIVERY_HINT;
|
use crate::mcp::REDELIVERY_HINT;
|
||||||
use crate::turn::{TurnError, TurnOutcome};
|
use crate::turn::{TurnError, TurnOutcome};
|
||||||
use crate::turn_stats::TurnStatRow;
|
use crate::turn_stats::TurnStatRow;
|
||||||
|
pub use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Assemble the per-turn wake prompt string. The role/tools/etc. live in the
|
/// Assemble the per-turn wake prompt string. The role/tools/etc. live in the
|
||||||
/// system prompt; this is just the wake signal body. `id` is the broker row
|
/// system prompt; this is just the wake signal body. `id` is the broker row
|
||||||
|
|
@ -46,13 +47,6 @@ pub fn format_wake_prompt(
|
||||||
|
|
||||||
/// Current time as a Unix timestamp (seconds). Returns 0 on any error.
|
/// Current time as a Unix timestamp (seconds). Returns 0 on any error.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn now_unix() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
|
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
|
||||||
/// columns; `outcome` and `bus` borrow for the duration of the call.
|
/// columns; `outcome` and `bus` borrow for the duration of the call.
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,7 @@ use rusqlite::{Connection, OpenFlags};
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
|
|
||||||
use hive_sh4re::ReminderStats;
|
use hive_sh4re::ReminderStats;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Window param accepted by `/api/stats?window=`. Each maps to a
|
/// Window param accepted by `/api/stats?window=`. Each maps to a
|
||||||
/// total span + the bucket width used to roll up trend series.
|
/// total span + the bucket width used to roll up trend series.
|
||||||
|
|
@ -208,7 +209,7 @@ fn default_path() -> PathBuf {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn empty_snapshot(window: Window) -> Snapshot {
|
fn empty_snapshot(window: Window) -> Snapshot {
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
let from = now - window.span_secs();
|
let from = now - window.span_secs();
|
||||||
let buckets = fill_buckets(from, now, window.bucket_secs(), &HashMap::new());
|
let buckets = fill_buckets(from, now, window.bucket_secs(), &HashMap::new());
|
||||||
Snapshot {
|
Snapshot {
|
||||||
|
|
@ -239,7 +240,7 @@ fn snapshot(path: &Path, window: Window) -> Result<Snapshot> {
|
||||||
// matches hive-c0re's host-side reader (`hive_stats::read_agent`).
|
// matches hive-c0re's host-side reader (`hive_stats::read_agent`).
|
||||||
conn.busy_timeout(std::time::Duration::from_millis(500))
|
conn.busy_timeout(std::time::Duration::from_millis(500))
|
||||||
.with_context(|| format!("set busy_timeout on {}", path.display()))?;
|
.with_context(|| format!("set busy_timeout on {}", path.display()))?;
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
// Fixed windows look back a constant span; `all` starts at the earliest
|
// Fixed windows look back a constant span; `all` starts at the earliest
|
||||||
// recorded turn (`MIN(started_at)`, falling back to `now` on an empty
|
// recorded turn (`MIN(started_at)`, falling back to `now` on an empty
|
||||||
// table) and sizes its buckets adaptively from that span.
|
// table) and sizes its buckets adaptively from that span.
|
||||||
|
|
@ -567,11 +568,6 @@ fn u64_from_i64(v: i64) -> u64 {
|
||||||
u64::try_from(v).unwrap_or(0)
|
u64::try_from(v).unwrap_or(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_secs() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
@ -638,7 +634,7 @@ mod tests {
|
||||||
fn snapshot_aggregates_rows() {
|
fn snapshot_aggregates_rows() {
|
||||||
let db = tmp_db();
|
let db = tmp_db();
|
||||||
let _ = std::fs::remove_file(&db);
|
let _ = std::fs::remove_file(&db);
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
seed_db(
|
seed_db(
|
||||||
&db,
|
&db,
|
||||||
&[
|
&[
|
||||||
|
|
@ -727,7 +723,7 @@ mod tests {
|
||||||
fn bash_breakdown_empty_without_table() {
|
fn bash_breakdown_empty_without_table() {
|
||||||
let db = tmp_db();
|
let db = tmp_db();
|
||||||
let _ = std::fs::remove_file(&db);
|
let _ = std::fs::remove_file(&db);
|
||||||
seed_db(&db, &[(now_secs() - 100, 1000, "opus", "recv", "ok", "{}")]);
|
seed_db(&db, &[(now_unix() - 100, 1000, "opus", "recv", "ok", "{}")]);
|
||||||
let s = snapshot(&db, Window::Day).unwrap();
|
let s = snapshot(&db, Window::Day).unwrap();
|
||||||
assert!(s.bash_breakdown.is_empty());
|
assert!(s.bash_breakdown.is_empty());
|
||||||
}
|
}
|
||||||
|
|
@ -739,7 +735,7 @@ mod tests {
|
||||||
let db = tmp_db();
|
let db = tmp_db();
|
||||||
let _ = std::fs::remove_file(&db);
|
let _ = std::fs::remove_file(&db);
|
||||||
seed_db(&db, &[]);
|
seed_db(&db, &[]);
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
let conn = Connection::open(&db).unwrap();
|
let conn = Connection::open(&db).unwrap();
|
||||||
conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);")
|
conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);")
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
|
||||||
|
|
@ -15,9 +15,10 @@
|
||||||
//! the honest fix is to clean them up where they live.
|
//! the honest fix is to clean them up where they live.
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
use std::time::Duration;
|
||||||
|
|
||||||
use rusqlite::{Connection, Result, params};
|
use rusqlite::{Connection, Result, params};
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// How often the sweep runs.
|
/// How often the sweep runs.
|
||||||
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
|
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
|
||||||
|
|
@ -131,10 +132,3 @@ fn vacuum_events(path: &Path) -> Result<u64> {
|
||||||
Ok(u64::try_from(removed).unwrap_or(0))
|
Ok(u64::try_from(removed).unwrap_or(0))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -4,11 +4,11 @@
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::Mutex;
|
use std::sync::Mutex;
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
|
||||||
|
|
||||||
use anyhow::{Context, Result, bail};
|
use anyhow::{Context, Result, bail};
|
||||||
use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus};
|
use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus};
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
use rusqlite::{Connection, OptionalExtension, params};
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
const SCHEMA: &str = r"
|
const SCHEMA: &str = r"
|
||||||
CREATE TABLE IF NOT EXISTS approvals (
|
CREATE TABLE IF NOT EXISTS approvals (
|
||||||
|
|
@ -380,13 +380,6 @@ fn kind_from_str(s: &str) -> Result<ApprovalKind> {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
|
||||||
|
|
@ -21,13 +21,13 @@
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::{Arc, Mutex, OnceLock};
|
use std::sync::{Arc, Mutex, OnceLock};
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
|
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
use rusqlite::{Connection, params};
|
use rusqlite::{Connection, params};
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Process-singleton handle, set once at coordinator startup. Mirrors
|
/// Process-singleton handle, set once at coordinator startup. Mirrors
|
||||||
/// `build_logs::GLOBAL` — lets recording sites write without threading an
|
/// `build_logs::GLOBAL` — lets recording sites write without threading an
|
||||||
|
|
@ -148,7 +148,7 @@ impl AuditLog {
|
||||||
outcome: AuditOutcome,
|
outcome: AuditOutcome,
|
||||||
detail: Option<&str>,
|
detail: Option<&str>,
|
||||||
) -> Option<AuditEntry> {
|
) -> Option<AuditEntry> {
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
match conn.execute(
|
match conn.execute(
|
||||||
"INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail)
|
"INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail)
|
||||||
|
|
@ -215,7 +215,7 @@ impl AuditLog {
|
||||||
/// # Errors
|
/// # Errors
|
||||||
/// Returns an error if the `DELETE` query fails.
|
/// Returns an error if the `DELETE` query fails.
|
||||||
pub fn vacuum(&self) -> Result<u64> {
|
pub fn vacuum(&self) -> Result<u64> {
|
||||||
let cutoff = now_secs() - KEEP_SECS;
|
let cutoff = now_unix() - KEEP_SECS;
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
let removed = conn.execute("DELETE FROM audit_log WHERE ts_unix < ?1", params![cutoff])?;
|
let removed = conn.execute("DELETE FROM audit_log WHERE ts_unix < ?1", params![cutoff])?;
|
||||||
Ok(u64::try_from(removed).unwrap_or(0))
|
Ok(u64::try_from(removed).unwrap_or(0))
|
||||||
|
|
@ -259,13 +259,6 @@ fn row_to_entry(r: &rusqlite::Row) -> rusqlite::Result<AuditEntry> {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_secs() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
@ -326,7 +319,7 @@ mod tests {
|
||||||
let conn = db.conn.lock().unwrap();
|
let conn = db.conn.lock().unwrap();
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"UPDATE audit_log SET ts_unix = ?1",
|
"UPDATE audit_log SET ts_unix = ?1",
|
||||||
params![now_secs() - KEEP_SECS - 60],
|
params![now_unix() - KEEP_SECS - 60],
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -4,7 +4,6 @@
|
||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::Mutex;
|
use std::sync::Mutex;
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
|
|
||||||
|
|
@ -13,6 +12,7 @@ use hive_sh4re::{InboxRow, Message};
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
use rusqlite::{Connection, OptionalExtension, params};
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
const SCHEMA: &str = r"
|
const SCHEMA: &str = r"
|
||||||
CREATE TABLE IF NOT EXISTS messages (
|
CREATE TABLE IF NOT EXISTS messages (
|
||||||
|
|
@ -1136,13 +1136,6 @@ fn ensure_reminder_columns(conn: &Connection) -> Result<()> {
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
|
||||||
|
|
@ -5,12 +5,12 @@
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::{Arc, Mutex, OnceLock};
|
use std::sync::{Arc, Mutex, OnceLock};
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
use rusqlite::{Connection, OptionalExtension, params};
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Process-singleton handle, set once at coordinator startup. Lets
|
/// Process-singleton handle, set once at coordinator startup. Lets
|
||||||
/// the `lifecycle` module's `run` / `prebuild_toplevel` access the
|
/// the `lifecycle` module's `run` / `prebuild_toplevel` access the
|
||||||
|
|
@ -169,7 +169,7 @@ impl BuildLogs {
|
||||||
/// — the caller threads it through `append_stdout` / `append_stderr`
|
/// — the caller threads it through `append_stdout` / `append_stderr`
|
||||||
/// while the child runs and into `finish` once it exits.
|
/// while the child runs and into `finish` once it exits.
|
||||||
pub fn start(&self, agent: &str, kind: &str, cmdline: &str) -> Result<i64> {
|
pub fn start(&self, agent: &str, kind: &str, cmdline: &str) -> Result<i64> {
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT INTO build_logs (agent, kind, cmdline, started_at) VALUES (?1, ?2, ?3, ?4)",
|
"INSERT INTO build_logs (agent, kind, cmdline, started_at) VALUES (?1, ?2, ?3, ?4)",
|
||||||
|
|
@ -219,7 +219,7 @@ impl BuildLogs {
|
||||||
/// Finalize a build attempt. Sets `finished_at` to now and
|
/// Finalize a build attempt. Sets `finished_at` to now and
|
||||||
/// `status` to the terminal state. Best-effort.
|
/// `status` to the terminal state. Best-effort.
|
||||||
pub fn finish(&self, id: i64, status: BuildStatus) {
|
pub fn finish(&self, id: i64, status: BuildStatus) {
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
if let Err(e) = conn.execute(
|
if let Err(e) = conn.execute(
|
||||||
"UPDATE build_logs SET finished_at = ?1, status = ?2 WHERE id = ?3",
|
"UPDATE build_logs SET finished_at = ?1, status = ?2 WHERE id = ?3",
|
||||||
|
|
@ -374,7 +374,7 @@ impl BuildLogs {
|
||||||
/// a long-running build shouldn't disappear from its own log
|
/// a long-running build shouldn't disappear from its own log
|
||||||
/// viewer mid-stream.
|
/// viewer mid-stream.
|
||||||
pub fn vacuum(&self) -> Result<u64> {
|
pub fn vacuum(&self) -> Result<u64> {
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
let fail_cutoff = now - KEEP_FAIL_SECS;
|
let fail_cutoff = now - KEEP_FAIL_SECS;
|
||||||
let ok_cutoff = now - KEEP_OK_SECS;
|
let ok_cutoff = now - KEEP_OK_SECS;
|
||||||
|
|
@ -438,13 +438,6 @@ fn row_to_header(r: &rusqlite::Row) -> rusqlite::Result<BuildLogHeader> {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_secs() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
@ -533,7 +526,7 @@ mod tests {
|
||||||
// stays within KEEP_FAIL_SECS so it survives; old_fail goes
|
// stays within KEEP_FAIL_SECS so it survives; old_fail goes
|
||||||
// beyond; old_ok goes past KEEP_OK_SECS but inside
|
// beyond; old_ok goes past KEEP_OK_SECS but inside
|
||||||
// KEEP_FAIL_SECS — proves the per-status rule.
|
// KEEP_FAIL_SECS — proves the per-status rule.
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
{
|
{
|
||||||
let conn = db.conn.lock().unwrap();
|
let conn = db.conn.lock().unwrap();
|
||||||
conn.execute(
|
conn.execute(
|
||||||
|
|
|
||||||
|
|
@ -18,12 +18,13 @@
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
use std::time::Duration;
|
||||||
|
|
||||||
use rusqlite::{Connection, OpenFlags};
|
use rusqlite::{Connection, OpenFlags};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
use crate::coordinator::Coordinator;
|
use crate::coordinator::Coordinator;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Window accepted by `/api/stats-hive?window=`. Maps to a lookback
|
/// Window accepted by `/api/stats-hive?window=`. Maps to a lookback
|
||||||
/// span; the hive view is a flat rollup (no per-bucket trend — the
|
/// span; the hive view is a flat rollup (no per-bucket trend — the
|
||||||
|
|
@ -225,11 +226,6 @@ struct AgentAgg {
|
||||||
bash: HashMap<String, u64>,
|
bash: HashMap<String, u64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_secs() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
|
|
||||||
}
|
|
||||||
|
|
||||||
#[allow(
|
#[allow(
|
||||||
clippy::cast_sign_loss,
|
clippy::cast_sign_loss,
|
||||||
|
|
@ -322,7 +318,7 @@ fn read_bash_heads(conn: &Connection, from: i64) -> HashMap<String, u64> {
|
||||||
/// per-agent db is skipped (logged), never fatal.
|
/// per-agent db is skipped (logged), never fatal.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn hive_snapshot(window: Window, prices: &PriceTable) -> HiveStats {
|
pub fn hive_snapshot(window: Window, prices: &PriceTable) -> HiveStats {
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
// Fixed windows look back a constant span; `all` aggregates every
|
// Fixed windows look back a constant span; `all` aggregates every
|
||||||
// recorded turn (`from == 0`). The hive rollup isn't time-bucketed, so
|
// recorded turn (`from == 0`). The hive rollup isn't time-bucketed, so
|
||||||
// unlike the per-agent snapshot it needs no adaptive bucket sizing.
|
// unlike the per-agent snapshot it needs no adaptive bucket sizing.
|
||||||
|
|
@ -426,7 +422,7 @@ mod tests {
|
||||||
let conn = Connection::open_in_memory().unwrap();
|
let conn = Connection::open_in_memory().unwrap();
|
||||||
conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);")
|
conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);")
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let now = now_secs();
|
let now = now_unix();
|
||||||
for (ts, head) in [
|
for (ts, head) in [
|
||||||
(now - 100, "cargo"),
|
(now - 100, "cargo"),
|
||||||
(now - 200, "cargo"),
|
(now - 200, "cargo"),
|
||||||
|
|
|
||||||
|
|
@ -29,6 +29,7 @@ use std::collections::{HashMap, VecDeque};
|
||||||
use std::sync::Mutex;
|
use std::sync::Mutex;
|
||||||
|
|
||||||
use tokio::sync::Notify;
|
use tokio::sync::Notify;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
pub use model::{
|
pub use model::{
|
||||||
Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, PermPayload, Source, State, Template,
|
Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, PermPayload, Source, State, Template,
|
||||||
|
|
@ -566,11 +567,3 @@ impl JobQueue {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Current unix timestamp in seconds.
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -13,12 +13,12 @@
|
||||||
//! the dashboard uses, so the bottleneck would be json
|
//! the dashboard uses, so the bottleneck would be json
|
||||||
//! (de)serialisation, not the read.
|
//! (de)serialisation, not the read.
|
||||||
|
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
|
||||||
|
|
||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
use hive_sh4re::LooseEnd;
|
use hive_sh4re::LooseEnd;
|
||||||
|
|
||||||
use crate::coordinator::Coordinator;
|
use crate::coordinator::Coordinator;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Open threads pending against `agent`:
|
/// Open threads pending against `agent`:
|
||||||
/// - undelivered inbox messages this agent still owes itself a `recv`
|
/// - undelivered inbox messages this agent still owes itself a `recv`
|
||||||
|
|
@ -143,13 +143,6 @@ fn saturating_age(now: i64, then: i64) -> u64 {
|
||||||
u64::try_from(delta).unwrap_or(0)
|
u64::try_from(delta).unwrap_or(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
|
||||||
|
|
@ -11,13 +11,13 @@
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::Mutex;
|
use std::sync::Mutex;
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
|
||||||
|
|
||||||
use anyhow::{Context, Result, bail};
|
use anyhow::{Context, Result, bail};
|
||||||
|
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
use rusqlite::{Connection, OptionalExtension, params};
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
const SCHEMA: &str = r"
|
const SCHEMA: &str = r"
|
||||||
CREATE TABLE IF NOT EXISTS operator_questions (
|
CREATE TABLE IF NOT EXISTS operator_questions (
|
||||||
|
|
@ -272,10 +272,3 @@ fn row_to_question(row: &rusqlite::Row<'_>) -> rusqlite::Result<OpQuestion> {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
SystemTime::now()
|
|
||||||
.duration_since(UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,7 @@ use std::sync::Mutex;
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
use rusqlite::{Connection, OptionalExtension, params};
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
const SCHEMA: &str = "
|
const SCHEMA: &str = "
|
||||||
CREATE TABLE IF NOT EXISTS agent_power (
|
CREATE TABLE IF NOT EXISTS agent_power (
|
||||||
|
|
@ -128,7 +129,7 @@ impl PowerStore {
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, ?3)
|
"INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, ?3)
|
||||||
ON CONFLICT(agent) DO UPDATE SET wanted = ?2, updated_at = ?3",
|
ON CONFLICT(agent) DO UPDATE SET wanted = ?2, updated_at = ?3",
|
||||||
params![agent, wanted.as_str(), now_secs()],
|
params![agent, wanted.as_str(), now_unix()],
|
||||||
)
|
)
|
||||||
.context("upsert agent_power")?;
|
.context("upsert agent_power")?;
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|
@ -157,13 +158,6 @@ impl PowerStore {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_secs() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
|
||||||
|
|
@ -18,6 +18,7 @@ use std::sync::Mutex;
|
||||||
use anyhow::{Context, Result, bail};
|
use anyhow::{Context, Result, bail};
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
use rusqlite::{Connection, OptionalExtension, params};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Typed error returned by [`ScheduledPrompts::pause`] and
|
/// Typed error returned by [`ScheduledPrompts::pause`] and
|
||||||
/// [`ScheduledPrompts::resume`] when the target row does not exist or
|
/// [`ScheduledPrompts::resume`] when the target row does not exist or
|
||||||
|
|
@ -674,13 +675,6 @@ fn load_targets(conn: &Connection, schedule_id: i64) -> Result<Vec<ScheduleTarge
|
||||||
Ok(out)
|
Ok(out)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
|
||||||
|
|
@ -10,6 +10,7 @@ use hive_sh4re::Message;
|
||||||
|
|
||||||
use crate::coordinator::Coordinator;
|
use crate::coordinator::Coordinator;
|
||||||
use crate::scheduled_prompts::Schedule;
|
use crate::scheduled_prompts::Schedule;
|
||||||
|
use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
||||||
/// Per-tick cap. Each schedule fires once per tick at most;
|
/// Per-tick cap. Each schedule fires once per tick at most;
|
||||||
/// 100/tick × 5s tick = sustained throughput cap of ~20/sec,
|
/// 100/tick × 5s tick = sustained throughput cap of ~20/sec,
|
||||||
|
|
@ -236,13 +237,6 @@ fn notify_operator_missing_target(coord: &Coordinator, schedule: &Schedule, targ
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Per-target outcome counts for one `fire_now` invocation.
|
/// Per-target outcome counts for one `fire_now` invocation.
|
||||||
/// Returned to the operator so the dashboard can render
|
/// Returned to the operator so the dashboard can render
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,19 @@ pub fn from_secs(secs: i64) -> DateTime<Utc> {
|
||||||
DateTime::<Utc>::from_timestamp(secs, 0).unwrap_or_default()
|
DateTime::<Utc>::from_timestamp(secs, 0).unwrap_or_default()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Current unix timestamp in seconds — the single definition behind
|
||||||
|
/// every store's `created_at` / `sent_at` / … stamp (this module owns
|
||||||
|
/// the epoch-seconds convention; a dozen local copies of this fn used
|
||||||
|
/// to float around both binaries). Clamps to 0 on a pre-epoch clock.
|
||||||
|
#[must_use]
|
||||||
|
pub fn now_unix() -> i64 {
|
||||||
|
std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.ok()
|
||||||
|
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
||||||
|
.unwrap_or(0)
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue