use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use serde::{Deserialize, Serialize};
use crate::Engine;
use crate::binding::{Binding, BuildMode};
use crate::pipeline::IngestTrigger;
use crate::pipeline_store::BindingConfigs;
use super::cursor::{source_moved, source_moved_since};
use super::findings::current_findings;
use super::resolve::{ResolvedIngest, resolve_binding_run};
pub const MAX_SKIP_LEVEL: u32 = 10;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum OperationKind {
Build,
Sync,
Verify,
}
impl OperationKind {
pub const ALL: [OperationKind; 3] = [
OperationKind::Build,
OperationKind::Sync,
OperationKind::Verify,
];
pub fn as_wire(&self) -> &'static str {
match self {
OperationKind::Build => "build",
OperationKind::Sync => "sync",
OperationKind::Verify => "verify",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OperationFilter {
Only(OperationKind),
Any,
}
impl OperationFilter {
fn admits(self, op: OperationKind) -> bool {
match self {
OperationFilter::Only(only) => only == op,
OperationFilter::Any => true,
}
}
}
fn pair_key(binding_id: &str, op: OperationKind) -> String {
format!("{binding_id}#{}", op.as_wire())
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackoffEntry {
#[serde(default)]
pub skip_remaining: u32,
#[serde(default)]
pub skip_level: u32,
#[serde(default)]
pub snapshot: String,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Cursor {
#[serde(default)]
pub last: Option<String>,
}
pub fn apply_backoff(entry: &mut BackoffEntry, current: &str) -> bool {
if !entry.snapshot.is_empty() && current != entry.snapshot {
entry.skip_remaining = 0;
entry.skip_level = 0;
entry.snapshot = current.to_string();
return false;
}
if entry.skip_remaining > 0 {
entry.skip_remaining -= 1;
return true;
}
if !entry.snapshot.is_empty() && current == entry.snapshot {
entry.skip_level = (entry.skip_level + 1).min(MAX_SKIP_LEVEL);
entry.skip_remaining = entry.skip_level;
}
entry.snapshot = current.to_string();
false
}
pub fn should_skip(
mode: BuildMode,
source_moved: bool,
entry: &mut BackoffEntry,
current: &str,
) -> bool {
match mode {
BuildMode::OneShot => return false,
BuildMode::Discovery => {}
}
if source_moved {
return false;
}
apply_backoff(entry, current)
}
fn read_json<T: Default + for<'de> Deserialize<'de>>(cache_root: &Path, name: &str) -> T {
std::fs::read(cache_root.join(name))
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
.unwrap_or_default()
}
fn write_json<T: Serialize>(cache_root: &Path, name: &str, value: &T) {
let _ = std::fs::create_dir_all(cache_root);
if let Ok(bytes) = serde_json::to_vec(value) {
let _ = std::fs::write(cache_root.join(name), bytes);
}
}
fn read_one_shot_runs(cache_root: &Path) -> BTreeSet<String> {
let map: BTreeMap<String, bool> = read_json(cache_root, "ingest-one-shot-runs.json");
map.into_iter()
.filter(|(_, v)| *v)
.map(|(k, _)| k)
.collect()
}
pub fn select_next_due(
engine: &Engine,
workspace_root: &Path,
configs: &BindingConfigs,
) -> Option<String> {
select_next_due_operation(
engine,
workspace_root,
configs,
OperationFilter::Only(OperationKind::Build),
)
.map(|(name, _)| name)
}
struct Pair<'a> {
key: String,
ingest: ResolvedIngest,
binding: &'a Binding,
op: OperationKind,
}
fn declared_for_loop(binding: &Binding, op: OperationKind) -> bool {
match op {
OperationKind::Build => binding
.operations
.build
.as_ref()
.is_some_and(|b| b.trigger == IngestTrigger::Loop),
OperationKind::Sync => binding
.operations
.sync
.as_ref()
.is_some_and(|s| s.trigger == IngestTrigger::Loop),
OperationKind::Verify => binding
.operations
.verify
.as_ref()
.is_some_and(|v| v.trigger == IngestTrigger::Loop),
}
}
fn operation_due(engine: &Engine, workspace_root: &Path, pair: &Pair<'_>) -> bool {
match pair.op {
OperationKind::Build => true,
OperationKind::Sync => {
source_moved(engine, &pair.ingest, workspace_root)
|| current_findings(engine, workspace_root, pair.binding, &pair.ingest)
.map(|(_key, findings)| !findings.is_empty())
.unwrap_or(false)
}
OperationKind::Verify => {
source_moved_since(engine, &pair.ingest, workspace_root, "verified", true)
}
}
}
pub fn select_next_due_operation(
engine: &Engine,
workspace_root: &Path,
configs: &BindingConfigs,
filter: OperationFilter,
) -> Option<(String, OperationKind)> {
let cache_root = workspace_root.join(".memstead.cache").join("ingest");
let one_shot_ran = read_one_shot_runs(&cache_root);
let mut eligible: Vec<Pair<'_>> = Vec::new();
for record in &configs.bindings {
let binding_id = format!("{}/{}", record.mem, record.name);
let Ok(ingest) = resolve_binding_run(&binding_id, &record.config) else {
continue;
};
for op in OperationKind::ALL {
if !filter.admits(op) || !declared_for_loop(&record.config, op) {
continue;
}
if op == OperationKind::Build
&& ingest.mode == BuildMode::OneShot
&& one_shot_ran.contains(&ingest.name)
{
continue;
}
eligible.push(Pair {
key: pair_key(&ingest.name, op),
ingest: ingest.clone(),
binding: &record.config,
op,
});
}
}
eligible.sort_by(|a, b| a.key.cmp(&b.key));
let n = eligible.len();
if n == 0 {
return None;
}
let mut cursor: Cursor = read_json(&cache_root, "ingest-cursor.json");
let start = cursor
.last
.as_ref()
.and_then(|last| eligible.iter().position(|p| &p.key == last))
.map_or(0, |i| (i + 1) % n);
cursor.last = Some(eligible[start].key.clone());
write_json(&cache_root, "ingest-cursor.json", &cursor);
let mut backoff: BTreeMap<String, BackoffEntry> = read_json(&cache_root, "ingest-backoff.json");
backoff.retain(|k, _| k.contains('#'));
let mut selected = None;
for offset in 0..n {
let pair = &eligible[(start + offset) % n];
if !operation_due(engine, workspace_root, pair) {
continue;
}
let current = engine
.mem_head_sha(&pair.ingest.destination_mem)
.ok()
.flatten()
.unwrap_or_default();
let (mode, moved) = match pair.op {
OperationKind::Build => (
pair.ingest.mode,
source_moved(engine, &pair.ingest, workspace_root),
),
OperationKind::Sync | OperationKind::Verify => (BuildMode::Discovery, false),
};
let entry = backoff.entry(pair.key.clone()).or_default();
if !should_skip(mode, moved, entry, ¤t) {
selected = Some((pair.ingest.name.clone(), pair.op));
break;
}
}
write_json(&cache_root, "ingest-backoff.json", &backoff);
selected
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn backoff_ramps_and_resets() {
let mut e = BackoffEntry::default();
assert!(!apply_backoff(&mut e, "sha1"));
assert_eq!(e.snapshot, "sha1");
assert_eq!(e.skip_level, 0);
assert!(!apply_backoff(&mut e, "sha1"));
assert_eq!(e.skip_level, 1);
assert_eq!(e.skip_remaining, 1);
assert!(apply_backoff(&mut e, "sha1"));
assert_eq!(e.skip_remaining, 0);
assert!(!apply_backoff(&mut e, "sha1"));
assert_eq!(e.skip_level, 2);
assert_eq!(e.skip_remaining, 2);
assert!(!apply_backoff(&mut e, "sha2"));
assert_eq!(e.skip_level, 0);
assert_eq!(e.skip_remaining, 0);
assert_eq!(e.snapshot, "sha2");
}
#[test]
fn backoff_caps_at_max_level() {
let mut e = BackoffEntry {
skip_level: MAX_SKIP_LEVEL,
skip_remaining: 0,
snapshot: "s".to_string(),
};
assert!(!apply_backoff(&mut e, "s")); assert_eq!(e.skip_level, MAX_SKIP_LEVEL, "capped");
assert_eq!(e.skip_remaining, MAX_SKIP_LEVEL);
}
#[test]
fn should_skip_honours_mode_and_source_movement() {
let mut e = BackoffEntry {
skip_remaining: 3,
skip_level: 3,
snapshot: "s".to_string(),
};
assert!(!should_skip(BuildMode::OneShot, false, &mut e.clone(), "s"));
let mut e2 = e.clone();
assert!(!should_skip(BuildMode::Discovery, true, &mut e2, "s"));
assert_eq!(e2.skip_remaining, 3, "moved source does not touch backoff");
assert!(should_skip(BuildMode::Discovery, false, &mut e, "s"));
}
use crate::binding::{
BINDING_VERSION, BuildOperation, Operations, SyncOperation, VerifyOperation, hash_binding,
};
use crate::pipeline::{MediumType, PatternEntry, PatternMode, Source};
use crate::pipeline_store::MemPipelineRecord;
use super::super::findings::{
Finding, FindingClass, FindingKey, FindingTarget, FindingsStore, write_findings_store,
};
fn empty_engine() -> Engine {
Engine::from_mounts(Vec::new()).unwrap()
}
fn binding_with(operations: Operations) -> Binding {
Binding {
version: BINDING_VERSION,
intent: None,
sources: Vec::new(),
reference_mems: Vec::new(),
destination_mem: "m".to_string(),
deny_paths: Vec::new(),
coverage_semantics: None,
rules: None,
prune: None,
operations,
}
}
fn build_op(trigger: IngestTrigger) -> BuildOperation {
BuildOperation {
mode: BuildMode::Discovery,
trigger,
batch_size: 20,
post_actions: None,
}
}
fn record(name: &str, config: Binding) -> MemPipelineRecord<Binding> {
MemPipelineRecord {
mem: "m".to_string(),
name: name.to_string(),
config,
}
}
fn configs_of(bindings: Vec<MemPipelineRecord<Binding>>) -> BindingConfigs {
BindingConfigs {
bindings,
quarantined: Vec::new(),
}
}
#[test]
fn eligibility_requires_block_and_loop_trigger() {
let ws = tempfile::tempdir().unwrap();
let engine = empty_engine();
let configs = configs_of(vec![
record(
"a",
binding_with(Operations {
build: Some(build_op(IngestTrigger::Loop)),
sync: None,
verify: None,
}),
),
record(
"b",
binding_with(Operations {
build: Some(build_op(IngestTrigger::Manual)),
sync: None,
verify: None,
}),
),
record(
"c",
binding_with(Operations {
build: None,
sync: Some(SyncOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
}),
verify: Some(VerifyOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
adjudication_cap: 50,
full_resync_every: 20,
}),
}),
),
]);
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Build)
),
Some(("m/a".to_string(), OperationKind::Build))
);
assert_eq!(
select_next_due_operation(&engine, ws.path(), &configs, OperationFilter::Any),
Some(("m/a".to_string(), OperationKind::Build))
);
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Sync)
),
None
);
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Verify)
),
None
);
}
#[test]
fn sync_pair_due_only_on_open_findings_when_source_unmoved() {
let ws = tempfile::tempdir().unwrap();
let engine = empty_engine();
let binding = binding_with(Operations {
build: None,
sync: Some(SyncOperation {
trigger: IngestTrigger::Loop,
batch_size: 20,
}),
verify: None,
});
let configs = configs_of(vec![record("s", binding.clone())]);
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Sync)
),
None
);
let key = FindingKey {
binding_hash: hash_binding(&binding),
source_head: String::new(),
};
let mut store = FindingsStore {
binding: "m/s".to_string(),
batches: Vec::new(),
};
store.record(key.clone(), "0".to_string(), Vec::new());
write_findings_store(ws.path(), "m", "s", &store).unwrap();
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Sync)
),
None
);
store.record(
key.clone(),
"1".to_string(),
vec![Finding {
key: key.clone(),
facet: "f".to_string(),
target: FindingTarget::Artifact {
artifact: "a.rs".to_string(),
},
class: FindingClass::Uncovered,
detail: "no anchor".to_string(),
created_at: "1".to_string(),
}],
);
write_findings_store(ws.path(), "m", "s", &store).unwrap();
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Sync)
),
Some(("m/s".to_string(), OperationKind::Sync))
);
let mut stale = FindingsStore {
binding: "m/s".to_string(),
batches: Vec::new(),
};
let stale_key = FindingKey {
binding_hash: "0000".to_string(),
source_head: "old".to_string(),
};
stale.record(
stale_key.clone(),
"1".to_string(),
vec![Finding {
key: stale_key,
facet: "f".to_string(),
target: FindingTarget::Artifact {
artifact: "a.rs".to_string(),
},
class: FindingClass::Uncovered,
detail: "stale".to_string(),
created_at: "1".to_string(),
}],
);
write_findings_store(ws.path(), "m", "s", &stale).unwrap();
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Sync)
),
None,
"superseded findings must not pull a sync into rotation"
);
}
fn configs_with_live_source(operations: Operations) -> BindingConfigs {
let mut binding = binding_with(operations);
binding.sources = vec![Source {
name: "f".to_string(),
medium_type: MediumType::Filesystem,
pointer: String::new(),
change_detection: Some("mtime".to_string()),
scope: vec![PatternEntry {
path: "**/*.rs".to_string(),
mode: PatternMode::Allow,
}],
engagement: None,
preparation: None,
}];
BindingConfigs {
bindings: vec![record("v", binding)],
quarantined: Vec::new(),
}
}
#[test]
fn verify_pair_due_when_never_verified_with_live_token() {
let ws = tempfile::tempdir().unwrap();
std::fs::write(ws.path().join("a.rs"), "x").unwrap();
let engine = empty_engine();
let verify_loop = Operations {
build: None,
sync: None,
verify: Some(VerifyOperation {
trigger: IngestTrigger::Loop,
batch_size: 20,
adjudication_cap: 50,
full_resync_every: 20,
}),
};
let configs = configs_with_live_source(verify_loop.clone());
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Verify)
),
Some(("m/v".to_string(), OperationKind::Verify))
);
let mut no_signal = configs_with_live_source(verify_loop);
no_signal.bindings[0].config.sources[0].scope.clear();
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&no_signal,
OperationFilter::Only(OperationKind::Verify)
),
None
);
}
#[test]
fn any_filter_rotates_across_pairs() {
let ws = tempfile::tempdir().unwrap();
std::fs::write(ws.path().join("a.rs"), "x").unwrap();
let engine = empty_engine();
let mut configs = configs_with_live_source(Operations {
build: None,
sync: None,
verify: Some(VerifyOperation {
trigger: IngestTrigger::Loop,
batch_size: 20,
adjudication_cap: 50,
full_resync_every: 20,
}),
});
configs.bindings.push(record(
"a",
binding_with(Operations {
build: Some(build_op(IngestTrigger::Loop)),
sync: None,
verify: None,
}),
));
let next = || {
select_next_due_operation(&engine, ws.path(), &configs, OperationFilter::Any).unwrap()
};
assert_eq!(next(), ("m/a".to_string(), OperationKind::Build));
assert_eq!(next(), ("m/v".to_string(), OperationKind::Verify));
assert_eq!(next(), ("m/a".to_string(), OperationKind::Build));
}
#[test]
fn legacy_single_key_backoff_entries_are_discarded() {
let ws = tempfile::tempdir().unwrap();
let engine = empty_engine();
let cache_root = ws.path().join(".memstead.cache").join("ingest");
std::fs::create_dir_all(&cache_root).unwrap();
let legacy: BTreeMap<String, BackoffEntry> = [(
"m/a".to_string(),
BackoffEntry {
skip_remaining: 5,
skip_level: 5,
snapshot: "s".to_string(),
},
)]
.into();
std::fs::write(
cache_root.join("ingest-backoff.json"),
serde_json::to_vec(&legacy).unwrap(),
)
.unwrap();
let configs = configs_of(vec![record(
"a",
binding_with(Operations {
build: Some(build_op(IngestTrigger::Loop)),
sync: None,
verify: None,
}),
)]);
assert_eq!(
select_next_due_operation(
&engine,
ws.path(),
&configs,
OperationFilter::Only(OperationKind::Build)
),
Some(("m/a".to_string(), OperationKind::Build)),
"a legacy entry's pending skips are discarded, not honoured"
);
let rewritten: BTreeMap<String, BackoffEntry> =
serde_json::from_slice(&std::fs::read(cache_root.join("ingest-backoff.json")).unwrap())
.unwrap();
assert!(!rewritten.contains_key("m/a"), "legacy key pruned");
assert!(rewritten.contains_key("m/a#build"), "pair key written");
}
}