1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
//! The perception offload-link sidecar: the drone's live "a workstation is
//! paired + reachable, and I am offloading to it" state.
//!
//! The offload reconciler (on the drone) writes this file each tick; the status
//! surfaces read it staleness-gated and feed a REAL `compute_node_paired` /
//! `bearer_acceptable` into [`crate`]'s tier decision, so an NPU-less drone with
//! a reachable workstation reports (and runs) tier `offload` instead of the
//! hardcoded "no node" default. Absent / stale ⇒ no offload link ⇒ the drone
//! reports `none` (honest: never a fabricated paired node).
//!
//! Producer: the `ados-cloud` offload reconciler. Consumers: `ados-control`
//! (`/api/status`) and `ados-cloud` (the cloud heartbeat). Single writer, many
//! readers — the same sidecar shape as compute-heartbeat / atlas-forward.
use std::path::Path;
use serde::{Deserialize, Serialize};
/// The offload-link sidecar path (tmpfs, cleared on boot).
pub const OFFLOAD_LINK_SIDECAR: &str = "/run/ados/offload-link.json";
/// The sidecar schema version, stamped on write and checked (warn-only) on read.
pub const OFFLOAD_LINK_SIDECAR_VERSION: u16 = 1;
/// A link file not re-written within this window is treated as absent, so a dead
/// / hung reconciler (whose tmpfs file persists) never keeps a drone reporting a
/// frozen `offload` tier after the workstation is gone. 4x
/// the reconciler's ~5 s tick.
pub const OFFLOAD_LINK_STALE_MS: i64 = 20_000;
/// The drone's live offload-link state, written by the reconciler each tick.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OffloadLink {
/// The sidecar schema version (absent ⇒ `0` from an older writer).
#[serde(default)]
pub version: u16,
/// The reconciler's write time (epoch ms); absent/stale ⇒ treated as gone.
pub generated_at_ms: Option<i64>,
/// A `profile=workstation` node resolved + is reachable this tick.
#[serde(default)]
pub paired: bool,
/// The LAN bearer to that node is healthy enough to offload over.
#[serde(default)]
pub bearer_acceptable: bool,
/// The node's job-API address (`host:port`) the drone is offloading to.
pub target: Option<String>,
/// The node's device id, when discovery reported one.
pub device_id: Option<String>,
/// The detector model id the offload session runs (labels the source).
pub model_id: Option<String>,
/// The plugin that advertised this link, when a plugin did. While the link
/// is fresh no other plugin may replace it.
#[serde(default)]
pub owner: Option<String>,
}
impl OffloadLink {
/// A link stamped at the current schema version + `now_ms` write time. The
/// reconciler builds one of these each tick and writes it; a not-paired tick
/// writes `paired: false` (an honest "no link right now") rather than leaving
/// a stale paired file behind.
pub fn stamped(
paired: bool,
bearer_acceptable: bool,
target: Option<String>,
device_id: Option<String>,
model_id: Option<String>,
now_ms: i64,
) -> Self {
Self {
version: OFFLOAD_LINK_SIDECAR_VERSION,
generated_at_ms: Some(now_ms),
paired,
bearer_acceptable,
target,
device_id,
model_id,
owner: None,
}
}
/// This link, recorded as advertised by plugin `owner`.
pub fn owned_by(mut self, owner: &str) -> Self {
self.owner = Some(owner.to_string());
self
}
/// Whether this link should count as an offload path for the tier decision:
/// a paired node on an acceptable bearer. (The staleness gate is applied by
/// the reader, so a link this method sees is already fresh.)
pub fn is_offload_path(&self) -> bool {
self.paired && self.bearer_acceptable
}
}
/// Read + parse the offload-link sidecar at `path`, or `None` when it is absent,
/// unparseable, missing its write-time, or STALE (older than
/// [`OFFLOAD_LINK_STALE_MS`] at `now_ms`). `now_ms` is injected so the staleness
/// gate is testable without touching the clock.
pub fn read_offload_link_from(path: &Path, now_ms: i64) -> Option<OffloadLink> {
let text = std::fs::read_to_string(path).ok()?;
let link: OffloadLink = serde_json::from_str(&text).ok()?;
match link.generated_at_ms {
Some(gen) if now_ms.saturating_sub(gen) <= OFFLOAD_LINK_STALE_MS => {
// Warn (never reject) on a producer/reader version mismatch, then
// fold the link in anyway — the same best-effort drift signal the
// other sidecars use.
crate::sidecar::check_sidecar_version(
"offload-link",
link.version,
OFFLOAD_LINK_SIDECAR_VERSION,
);
Some(link)
}
_ => None,
}
}
/// Read the offload-link sidecar from the default path, staleness-gated at
/// `now_ms`.
pub fn read_offload_link(now_ms: i64) -> Option<OffloadLink> {
read_offload_link_from(Path::new(OFFLOAD_LINK_SIDECAR), now_ms)
}
/// Atomically write `link` to `path` (tmp + rename), creating the parent dir if
/// needed. The reconciler stamps the link ([`OffloadLink::stamped`]) before
/// calling this.
pub fn write_offload_link_to(path: &Path, link: &OffloadLink) -> std::io::Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let body = serde_json::to_vec(link).map_err(std::io::Error::other)?;
let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, body)?;
std::fs::rename(&tmp, path)
}
/// Atomically write `link` to the default sidecar path.
pub fn write_offload_link(link: &OffloadLink) -> std::io::Result<()> {
write_offload_link_to(Path::new(OFFLOAD_LINK_SIDECAR), link)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_fresh_paired_link_round_trips_and_reads_back() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("offload-link.json");
let now = 1_700_000_000_000i64;
let link = OffloadLink::stamped(
true,
true,
Some("192.168.1.5:8092".into()),
Some("dev-abc".into()),
Some("coco-yolov8n".into()),
now,
);
write_offload_link_to(&path, &link).unwrap();
// No leftover temp file.
assert!(!dir.path().join("offload-link.json.tmp").exists());
let back = read_offload_link_from(&path, now + 1_000).unwrap();
assert_eq!(back, link);
assert!(back.is_offload_path());
}
#[test]
fn a_stale_link_reads_as_absent() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("offload-link.json");
let gen = 1_700_000_000_000i64;
let link = OffloadLink::stamped(true, true, Some("h:8092".into()), None, None, gen);
write_offload_link_to(&path, &link).unwrap();
// Just inside the window ⇒ present; just past it ⇒ absent.
assert!(read_offload_link_from(&path, gen + OFFLOAD_LINK_STALE_MS).is_some());
assert!(read_offload_link_from(&path, gen + OFFLOAD_LINK_STALE_MS + 1).is_none());
}
#[test]
fn an_absent_file_reads_as_none() {
let p = Path::new("/run/ados/does-not-exist-offload-link.json");
assert!(read_offload_link_from(p, 1_700_000_000_000).is_none());
}
#[test]
fn a_not_paired_link_is_not_an_offload_path() {
let link = OffloadLink::stamped(false, false, None, None, None, 1);
assert!(!link.is_offload_path());
}
}