reifydb-flow 0.9.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

use std::{
	collections::{BTreeMap, HashMap},
	fmt::Debug,
	hash::Hash,
	marker::PhantomData,
	slice::from_ref,
};

use reifydb_codec::{
	key::encoded::{EncodedKey, IntoEncodedKey},
	row::operator::{OperatorState, decode},
};
use reifydb_core::{
	key::operator_state::{GroupId, GroupStateKey, IntoGroupStateKey},
	metrics::heap::HeapSize,
	state::{cache::StateCache, store::StateStore},
};
use reifydb_macro::operator_state;
use reifydb_value::{Result, reifydb_assertions};

use crate::window::{
	accumulator::WindowAccumulator,
	engine::{
		AccumulatorEvent, EmitKind, MetaHighWater, MetaKey, WindowResult, WindowStateKey,
		config::TumblingCarryConfig, meta_key_for, sweep_stale_meta, tumbling::TumblingBuckets,
	},
	span::{SlotSpan, WindowAnchor, WindowSpan},
};

#[operator_state]
#[derive(Debug, Clone)]
pub struct WindowEntry<C, Carry, Output> {
	span: WindowSpan<C>,
	carry_out: Option<Carry>,
	last_output: Option<Output>,
}

impl<C: HeapSize, Carry: HeapSize, Output: HeapSize> HeapSize for WindowEntry<C, Carry, Output> {
	fn heap_size(&self) -> usize {
		self.span.heap_size() + self.carry_out.heap_size() + self.last_output.heap_size()
	}
}

#[operator_state]
#[derive(Debug, Clone)]
pub struct CarryMeta<C, Carry, Output> {
	high_water: Option<C>,
	sealed_up_to: Option<C>,
	sealed_carry: Option<Carry>,
	windows: BTreeMap<C, WindowEntry<C, Carry, Output>>,
}

impl<C: HeapSize, Carry: HeapSize, Output: HeapSize> HeapSize for CarryMeta<C, Carry, Output> {
	fn heap_size(&self) -> usize {
		self.high_water.heap_size()
			+ self.sealed_up_to.heap_size()
			+ self.sealed_carry.heap_size()
			+ self.windows.heap_size()
	}
}

impl<C, Carry, Output> Default for CarryMeta<C, Carry, Output> {
	fn default() -> Self {
		Self {
			high_water: None,
			sealed_up_to: None,
			sealed_carry: None,
			windows: BTreeMap::new(),
		}
	}
}

impl<C: WindowAnchor, Carry, Output> MetaHighWater for CarryMeta<C, Carry, Output>
where
	Self: OperatorState,
{
	fn high_water_order(&self) -> Option<u64> {
		self.high_water.map(|hw| hw.order_key().to_order())
	}
}

type MetaLoaded<G, C, Carry, Output> = HashMap<G, CarryMeta<C, Carry, Output>>;
type SlotResolved = Vec<Option<(GroupId, EncodedKey)>>;

struct PendingCarry<C, Output> {
	group_id: GroupId,
	key: EncodedKey,
	span: WindowSpan<C>,
	value: Output,
	withdraw: bool,
}

pub struct TumblingCarryEngine<G, C: WindowAnchor, Accumulator, Carry, Output> {
	accumulators: StateCache<WindowStateKey, Accumulator>,
	meta: StateCache<MetaKey, CarryMeta<C, Carry, Output>>,
	meta_low_water: Option<u64>,
	retention: Option<SlotSpan<C>>,
	_pd: PhantomData<G>,
}

impl<G, C, Accumulator, Carry, Output> TumblingCarryEngine<G, C, Accumulator, Carry, Output>
where
	G: Clone + Eq + Ord + Hash + Debug,
	C: WindowAnchor + Hash,
	Accumulator: WindowAccumulator,
	Carry: Clone + Debug,
	Output: Clone + Debug,
	for<'a> &'a G: IntoEncodedKey,
	C: HeapSize,
	Carry: HeapSize,
	Output: HeapSize,
	CarryMeta<C, Carry, Output>: OperatorState,
{
	pub fn new(config: TumblingCarryConfig<C>) -> Self {
		Self {
			accumulators: StateCache::<WindowStateKey, Accumulator>::new(),
			meta: StateCache::<MetaKey, CarryMeta<C, Carry, Output>>::new(),
			meta_low_water: None,
			retention: config.retention(),
			_pd: PhantomData,
		}
	}

	pub fn expire_meta(&mut self, store: &mut dyn StateStore, threshold: u64) -> Result<usize> {
		sweep_stale_meta(store, &mut self.meta, threshold, &mut self.meta_low_water)
	}

	#[allow(clippy::too_many_arguments)]
	pub fn apply<K, NA, BO, CF>(
		&mut self,
		store: &mut dyn StateStore,
		buckets: TumblingBuckets<G, C, Accumulator::Contribution>,
		row_key: K,
		new_accumulator: NA,
		build_output: BO,
		carry_forward: CF,
	) -> Result<Vec<WindowResult<G, C, Output>>>
	where
		K: Fn(&G, C) -> EncodedKey,
		NA: Fn() -> Accumulator,
		BO: Fn(&G, WindowSpan<C>, &Accumulator::Output, Option<&Carry>) -> Option<Output>,
		CF: Fn(&Accumulator::Output, Option<&Carry>) -> Option<Carry>,
	{
		if buckets.is_empty() {
			return Ok(Vec::new());
		}
		let retention = self.retention;
		let mut meta_loaded = self.load_meta(store, &buckets)?;
		let slot_resolved = self.resolve_survivor_rows(store, &buckets, &meta_loaded, &row_key)?;

		let mut earliest_affected: HashMap<G, C> = HashMap::new();
		for (((group, span), events), slot_pre) in buckets.into_iter().zip(slot_resolved) {
			let entry = meta_loaded.entry(group.clone()).or_default();
			if matches!(entry.sealed_up_to, Some(s) if span.start <= s) {
				continue;
			}
			let slot_key = row_key(&group, span.start);
			let group_id = match &slot_pre {
				Some((gid, _)) => *gid,
				None => store.intern_groups(from_ref(&slot_key))?.into_iter().next().unwrap().0,
			};
			if !entry.windows.contains_key(&span.start) && slot_pre.is_none() {
				continue;
			}

			let mut accumulator: Accumulator = self
				.accumulators
				.get(store, &WindowStateKey::new(group_id, slot_key.clone()))?
				.unwrap_or_else(&new_accumulator);
			let mut changed = false;
			for event in events {
				match event {
					AccumulatorEvent::Add(c) => {
						accumulator.add(&c);
						changed = true;
					}
					AccumulatorEvent::Remove(c) => {
						if accumulator.is_empty() {
							continue;
						}
						accumulator.remove(&c);
						changed = true;
					}
				}
			}
			if !changed {
				continue;
			}
			self.accumulators.put(store, &WindowStateKey::new(group_id, slot_key), accumulator)?;

			entry.windows.entry(span.start).or_insert_with(|| WindowEntry {
				span,
				carry_out: None,
				last_output: None,
			});
			if entry.high_water.is_none_or(|hw| span.start > hw) {
				entry.high_water = Some(span.start);
			}

			let e = earliest_affected.entry(group).or_insert(span.start);
			if span.start < *e {
				*e = span.start;
			}
		}

		let mut results: Vec<WindowResult<G, C, Output>> = Vec::new();
		for (group, start) in earliest_affected {
			let meta = meta_loaded.get_mut(&group).expect("affected group has meta");

			let mut prev_carry: Option<Carry> = match meta.windows.range(..start).next_back() {
				Some((_, w)) => w.carry_out.clone(),
				None => meta.sealed_carry.clone(),
			};

			let coords: Vec<C> = meta.windows.range(start..).map(|(c, _)| *c).collect();
			let coord_keys: Vec<EncodedKey> = coords.iter().map(|coord| row_key(&group, *coord)).collect();
			let coord_groups = store.lookup_groups(&coord_keys)?;

			let mut emptied: Vec<C> = Vec::new();
			let mut pending: Vec<PendingCarry<C, Output>> = Vec::new();
			for ((coord, slot_key), coord_group) in coords.into_iter().zip(coord_keys).zip(coord_groups) {
				let span = meta.windows.get(&coord).expect("window entry present").span;
				let finalized = match coord_group {
					Some(coord_group) => self
						.accumulators
						.get(store, &WindowStateKey::new(coord_group, slot_key.clone()))?
						.and_then(|a| a.finalize())
						.map(|value| (coord_group, value)),
					None => None,
				};
				let emitted = finalized.as_ref().and_then(|(coord_group, value)| {
					build_output(&group, span, value, prev_carry.as_ref())
						.map(|out| (*coord_group, value, out))
				});
				match emitted {
					Some((coord_group, value, out)) => {
						let new_carry = carry_forward(value, prev_carry.as_ref());
						let w = meta.windows.get_mut(&coord).expect("window entry present");
						w.carry_out = new_carry.clone();
						w.last_output = Some(out.clone());
						if new_carry.is_some() {
							prev_carry = new_carry;
						}
						pending.push(PendingCarry {
							group_id: coord_group,
							key: slot_key,
							span,
							value: out,
							withdraw: false,
						});
					}
					None => {
						if let Some(prev) =
							meta.windows.get(&coord).and_then(|w| w.last_output.clone())
							&& let Some(coord_group) = coord_group
						{
							pending.push(PendingCarry {
								group_id: coord_group,
								key: slot_key,
								span,
								value: prev,
								withdraw: true,
							});
						}
						emptied.push(coord);
					}
				}
			}

			let pairs: Vec<(GroupId, EncodedKey)> =
				pending.iter().map(|p| (p.group_id, p.key.clone())).collect();
			let rows = store.get_or_create_row_numbers_for_pairs(&pairs)?;
			reifydb_assertions! {
				let requested = pairs.len();
				let returned = rows.len();
				assert!(
					returned == requested,
					"the identity batch must return one row per publishing window; a short batch makes \
					 the zip below drop the tail, so those windows publish nothing while their carry \
					 meta already advanced (requested={requested}, returned={returned})"
				);
			}
			for (emit, (row_number, is_new)) in pending.into_iter().zip(rows) {
				let kind = if emit.withdraw {
					store.remove_row_number(emit.group_id, &emit.key)?;
					EmitKind::Remove
				} else if is_new {
					EmitKind::Insert
				} else {
					EmitKind::Update
				};
				results.push(WindowResult {
					row_number,
					group: group.clone(),
					span: emit.span,
					value: emit.value,
					prior: None,
					kind,
				});
			}

			for coord in emptied {
				meta.windows.remove(&coord);
			}

			if let (Some(retention), Some(hw)) = (retention, meta.high_water) {
				let to_seal: Vec<C> = meta
					.windows
					.keys()
					.copied()
					.take_while(|first| hw.span_since(*first) > retention)
					.collect();
				let sealed_keys: Vec<EncodedKey> =
					to_seal.iter().map(|first| row_key(&group, *first)).collect();
				let sealed_groups = store.lookup_groups(&sealed_keys)?;
				for ((first, sealed_key), sealed_group) in
					to_seal.into_iter().zip(sealed_keys).zip(sealed_groups)
				{
					let carry_out = meta
						.windows
						.get(&first)
						.expect("sealed window entry present")
						.carry_out
						.clone();
					meta.windows.remove(&first);
					meta.sealed_up_to = Some(first);
					meta.sealed_carry = carry_out;
					if let Some(sealed_group) = sealed_group {
						self.accumulators.remove(
							store,
							&WindowStateKey::new(sealed_group, sealed_key.clone()),
						)?;
						store.remove_row_number(sealed_group, &sealed_key)?;
					}
				}
			}
		}

		self.persist_meta(store, meta_loaded)?;
		Ok(results)
	}

	fn load_meta(
		&mut self,
		store: &mut dyn StateStore,
		buckets: &TumblingBuckets<G, C, Accumulator::Contribution>,
	) -> Result<MetaLoaded<G, C, Carry, Output>> {
		let mut meta_loaded: MetaLoaded<G, C, Carry, Output> = HashMap::new();
		let mut by_key: HashMap<GroupStateKey, G> = HashMap::new();
		for (group, _) in buckets.keys() {
			if meta_loaded.contains_key(group) {
				continue;
			}
			meta_loaded.insert(group.clone(), CarryMeta::default());
			by_key.insert((&meta_key_for(group)).into_group_state_key(), group.clone());
		}
		let keys: Vec<GroupStateKey> = by_key.keys().cloned().collect();
		store.state_get_many_visit(&keys, &mut |key, bytes| {
			if let Some(group) = by_key.get(&key) {
				meta_loaded.insert(group.clone(), decode::<CarryMeta<C, Carry, Output>>(&bytes)?);
			}
			Ok(())
		})?;
		Ok(meta_loaded)
	}

	fn resolve_survivor_rows<K>(
		&mut self,
		store: &mut dyn StateStore,
		buckets: &TumblingBuckets<G, C, Accumulator::Contribution>,
		meta_loaded: &MetaLoaded<G, C, Carry, Output>,
		row_key: &K,
	) -> Result<SlotResolved>
	where
		K: Fn(&G, C) -> EncodedKey,
	{
		let mut survivor_keys: Vec<EncodedKey> = Vec::new();
		let mut slot_survives: Vec<bool> = Vec::with_capacity(buckets.len());
		for (group, span) in buckets.keys() {
			let meta = meta_loaded.get(group);
			let sealed = matches!(meta.and_then(|m| m.sealed_up_to), Some(s) if span.start <= s);
			let survives = !sealed;
			slot_survives.push(survives);
			if survives {
				survivor_keys.push(row_key(group, span.start));
			}
		}
		let interned = store.intern_groups(&survivor_keys)?;
		let resolved_rows: Vec<(GroupId, EncodedKey)> =
			survivor_keys.iter().cloned().zip(interned).map(|(key, (group, _))| (group, key)).collect();
		reifydb_assertions! {
			let survivors = survivor_keys.len();
			let resolved = resolved_rows.len();
			assert!(
				resolved == survivors,
				"intern_group must return exactly one group per survivor key; a short batch would \
				 leave a surviving slot with no resolved group, so the slot_resolved zip below pairs it with None \
				 and apply silently skips the slot instead of folding into the existing window state \
				 (survivor_keys={survivors}, resolved_rows={resolved})"
			);
		}
		let mut resolved_rows = resolved_rows.into_iter();
		Ok(slot_survives
			.into_iter()
			.map(|survives| {
				if survives {
					resolved_rows.next()
				} else {
					None
				}
			})
			.collect())
	}

	fn persist_meta(
		&mut self,
		store: &mut dyn StateStore,
		meta_loaded: MetaLoaded<G, C, Carry, Output>,
	) -> Result<()> {
		for (group, meta) in meta_loaded {
			self.meta.put(store, &meta_key_for(&group), meta)?;
		}
		Ok(())
	}
}

#[cfg(test)]
mod tests {
	use std::{collections::HashMap, ops::Bound};

	use reifydb_codec::{
		key::encoded::EncodedKeyRange,
		row::operator::{EncodedOperatorRow, decode},
	};
	use reifydb_core::{
		key::operator_state::{GroupStateKey, Keyspace, OperatorStateKey},
		state::store::{TimerKind, TimerStore},
	};
	use reifydb_value::{
		factory::time::{at_millis, millis},
		value::{datetime::DateTime, duration::Duration, row_number::RowNumber},
	};

	use super::*;
	use crate::{
		operator::state::seal::coord::Coord,
		window::{
			accumulator::invertible::retained_map::RetainedAccumulator, engine::config::WindowEngineConfig,
		},
	};

	// Allocates a distinct row number per key; the state.rs mock collapses every key onto row 1,
	// which would alias all window accumulators and defeat a storage-bound test.
	#[derive(Default)]
	struct CountingStore {
		data: HashMap<Vec<u8>, EncodedOperatorRow>,
		groups: HashMap<Vec<u8>, GroupId>,
		rows: HashMap<(GroupId, Vec<u8>), RowNumber>,
		next_row: u64,
	}

	impl CountingStore {
		fn keyspace_count(&self, keyspace: Keyspace) -> usize {
			self.data
				.keys()
				.filter(|k| {
					OperatorStateKey::decode_inner(k).is_some_and(|(_, found, _)| found == keyspace)
				})
				.count()
		}

		fn accumulator_count(&self) -> usize {
			// Meta and the expiry index share the store, so count only the accumulator keyspace.
			self.keyspace_count(Keyspace::ACCUMULATOR)
		}

		fn meta_entry_count(&self) -> usize {
			self.keyspace_count(Keyspace::WINDOW_META)
		}

		fn row_mapping_count(&self) -> usize {
			// One mapping is minted per (group, window), so this is what proves a sealed window
			// reclaims its own.
			self.rows.len()
		}

		fn drop_group_data_entries(&mut self) -> usize {
			// Phase-1 reclamation clears every data keyspace inside a real group but leaves the root group
			// alone; row-number mappings live outside `data` and survive it the same way production does.
			let keys: Vec<Vec<u8>> = self
				.data
				.keys()
				.filter(|k| {
					OperatorStateKey::decode_inner(k)
						.is_some_and(|(group, found, _)| !group.is_root() && found.is_data())
				})
				.cloned()
				.collect();
			for key in &keys {
				self.data.remove(key);
			}
			keys.len()
		}
	}

	impl TimerStore for CountingStore {
		fn arm_timer(&mut self, _due: DateTime, _kind: TimerKind, _key: &EncodedKey) -> Result<()> {
			unreachable!("the window engine never arms timers; only the shell above it does")
		}

		fn disarm_timer(&mut self, _due: DateTime, _kind: TimerKind, _key: &EncodedKey) -> Result<()> {
			unreachable!("the window engine never disarms timers; only the shell above it does")
		}

		fn flow_watermark(&mut self) -> Result<Option<DateTime>> {
			Ok(None)
		}
	}

	impl CountingStore {
		fn row_number_for(&mut self, group: GroupId, key: &EncodedKey) -> (RowNumber, bool) {
			let slot = (group, key.as_bytes().to_vec());
			if let Some(rn) = self.rows.get(&slot) {
				return (*rn, false);
			}
			self.next_row += 1;
			let rn = RowNumber(self.next_row);
			self.rows.insert(slot, rn);
			(rn, true)
		}
	}

	impl StateStore for CountingStore {
		fn intern_groups(&mut self, groups: &[EncodedKey]) -> Result<Vec<(GroupId, bool)>> {
			let mut interned = Vec::with_capacity(groups.len());
			for group in groups {
				let bytes = group.as_bytes().to_vec();
				match self.groups.get(&bytes) {
					Some(id) => interned.push((*id, false)),
					None => {
						let next = GroupId(self.groups.len() as u64 + GroupId::FIRST.0);
						self.groups.insert(bytes, next);
						interned.push((next, true));
					}
				}
			}
			Ok(interned)
		}

		fn lookup_groups(&mut self, groups: &[EncodedKey]) -> Result<Vec<Option<GroupId>>> {
			Ok(groups.iter().map(|group| self.groups.get(group.as_bytes()).copied()).collect())
		}

		fn state_get(&mut self, key: &GroupStateKey) -> Result<Option<EncodedOperatorRow>> {
			Ok(self.data.get(key.as_slice()).cloned())
		}
		fn state_get_many_visit(
			&mut self,
			keys: &[GroupStateKey],
			visit: &mut dyn FnMut(GroupStateKey, EncodedOperatorRow) -> Result<()>,
		) -> Result<()> {
			for key in keys {
				if let Some(b) = self.data.get(key.as_slice()) {
					visit(key.clone(), b.clone())?;
				}
			}
			Ok(())
		}
		fn state_set(&mut self, key: &GroupStateKey, payload: EncodedOperatorRow) -> Result<()> {
			self.data.insert(key.as_slice().to_vec(), payload);
			Ok(())
		}
		fn state_remove(&mut self, key: &GroupStateKey) -> Result<()> {
			self.data.remove(key.as_slice());
			Ok(())
		}
		fn state_range_visit(
			&mut self,
			range: EncodedKeyRange,
			limit: Option<usize>,
			visit: &mut dyn FnMut(GroupStateKey, EncodedOperatorRow) -> Result<()>,
		) -> Result<()> {
			let after_start = |k: &[u8]| match &range.start {
				Bound::Included(s) => k >= s.as_bytes(),
				Bound::Excluded(s) => k > s.as_bytes(),
				Bound::Unbounded => true,
			};
			let before_end = |k: &[u8]| match &range.end {
				Bound::Included(e) => k <= e.as_bytes(),
				Bound::Excluded(e) => k < e.as_bytes(),
				Bound::Unbounded => true,
			};
			let mut matched: Vec<(Vec<u8>, EncodedOperatorRow)> = self
				.data
				.iter()
				.filter(|(k, _)| after_start(k) && before_end(k))
				.map(|(k, v)| (k.clone(), v.clone()))
				.collect();
			matched.sort_by(|a, b| a.0.cmp(&b.0));
			if let Some(limit) = limit {
				matched.truncate(limit);
			}
			for (k, b) in matched {
				let k = GroupStateKey::from_framed(EncodedKey::new(k))
					.expect("fake store holds an unframed state key");
				visit(k, b)?;
			}
			Ok(())
		}
		fn get_or_create_row_numbers(
			&mut self,
			group: GroupId,
			keys: &[EncodedKey],
		) -> Result<Vec<(RowNumber, bool)>> {
			Ok(keys.iter().map(|key| self.row_number_for(group, key)).collect())
		}
		fn get_or_create_row_numbers_for_pairs(
			&mut self,
			pairs: &[(GroupId, EncodedKey)],
		) -> Result<Vec<(RowNumber, bool)>> {
			Ok(pairs.iter().map(|(group, key)| self.row_number_for(*group, key)).collect())
		}
		fn remove_row_number(&mut self, group: GroupId, key: &EncodedKey) -> Result<()> {
			self.rows.remove(&(group, key.as_bytes().to_vec()));
			Ok(())
		}
		fn written_at(&self) -> DateTime {
			DateTime::EPOCH
		}
	}

	type Engine = TumblingCarryEngine<String, DateTime, RetainedAccumulator<u64, f64>, f64, f64>;

	const WINDOW: u64 = 60;

	fn order(millis: u64) -> u64 {
		at_millis(millis).to_order()
	}

	fn carry_config(retention: Option<Duration>) -> TumblingCarryConfig<DateTime> {
		TumblingCarryConfig::builder(WindowEngineConfig::builder().build()).retention(retention).build()
	}

	fn feed(engine: &mut Engine, store: &mut CountingStore, ws: DateTime, price: f64) {
		// One event per batch, so the high-water mark advances one window per call.
		let _ = feed_group(engine, store, "BTC", ws, price);
	}

	// Distinct groups get distinct accumulator rows, which is what the eviction test needs to
	// overflow the 8-slot accumulator cache.
	fn feed_group(
		engine: &mut Engine,
		store: &mut CountingStore,
		group: &str,
		ws: DateTime,
		price: f64,
	) -> Vec<WindowResult<String, DateTime, f64>> {
		let mut buckets: TumblingBuckets<String, DateTime, (u64, f64)> = BTreeMap::new();
		let span = WindowSpan::for_coord(ws, millis(WINDOW));
		buckets.insert((group.to_string(), span), vec![AccumulatorEvent::Add((ws.to_order(), price))]);
		engine.apply(
			store,
			buckets,
			|g: &String, w: DateTime| EncodedKey::builder().str(g).u64(w.to_order()).build(),
			RetainedAccumulator::<u64, f64>::default,
			|_g: &String, _s: WindowSpan<DateTime>, v: &BTreeMap<u64, f64>, _p: Option<&f64>| {
				(!v.is_empty()).then(|| v.values().sum::<f64>())
			},
			|v: &BTreeMap<u64, f64>, _p: Option<&f64>| v.last_key_value().map(|(_, val)| *val),
		)
		.expect("apply")
	}

	#[test]
	fn retention_seals_old_windows_and_reclaims_accumulator_rows() {
		// With a 2-window retention horizon, older windows must seal to the O(1) carry scalar and
		// drop their accumulator row, so the live row count stays bounded by the horizon rather
		// than growing with the number of windows seen.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(Some(millis(2 * WINDOW))));
		for i in 0..60u64 {
			feed(&mut engine, &mut store, at_millis(i * WINDOW), i as f64);
		}
		assert!(
			store.accumulator_count() <= 4,
			"sealed windows must reclaim their accumulator rows; found {} live rows after 60 windows",
			store.accumulator_count()
		);
	}

	#[test]
	fn retention_seals_old_windows_and_reclaims_row_number_mappings() {
		// The per-(group, window) mapping is keyed by row_key, not row_number, so accumulator
		// eviction does not reclaim it. Sealing past retention must drop it alongside the
		// accumulator or the mapping keyspace grows per window forever.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(Some(millis(2 * WINDOW))));
		for i in 0..60u64 {
			feed(&mut engine, &mut store, at_millis(i * WINDOW), i as f64);
		}
		assert!(
			store.row_mapping_count() <= 4,
			"sealed windows must reclaim their row-number mappings; found {} live mappings after 60 windows",
			store.row_mapping_count()
		);
	}

	#[test]
	fn a_window_whose_state_was_reclaimed_updates_its_row_rather_than_inserting_a_second() {
		// The data phase takes the accumulator, the carry meta and last_output at once while the
		// mapping stays addressable, so what comes back has no memory of what it published but the
		// sink still holds that row. A second insert would lay a duplicate row over a live one.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(None));
		let published = feed_group(&mut engine, &mut store, "BTC", at_millis(0), 5.0);
		assert_eq!(published.len(), 1);
		assert!(matches!(published[0].kind, EmitKind::Insert), "precondition: the window publishes once");

		assert!(store.drop_group_data_entries() > 0, "precondition: the sweep must have erased something");
		assert_eq!(store.row_mapping_count(), 1, "precondition: the identity half must survive the data phase");

		let mut engine = Engine::new(carry_config(None));
		let republished = feed_group(&mut engine, &mut store, "BTC", at_millis(0), 3.0);

		assert_eq!(republished.len(), 1);
		assert_eq!(
			republished[0].kind,
			EmitKind::Update,
			"the published row survived the sweep, so this is an update and not a second insert"
		);
		assert_eq!(
			republished[0].row_number, published[0].row_number,
			"the woken window keeps the row it published"
		);
	}

	#[test]
	fn every_successive_window_emits_its_own_result() {
		// Production symptom: a `apply twap { window_duration: '1m' }` ladder over an advancing
		// event-time stream published output for the first window only and then froze, while its
		// source view kept advancing. Every existing test here feeds many windows but only asserts
		// on reclamation counts, so a driver that stops emitting after the first window passes them
		// all. Each successive window carries its own events, so each must publish its own row.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(None));
		let mut emitted_windows = Vec::new();
		for i in 0..5u64 {
			let out = feed_group(&mut engine, &mut store, "BTC", at_millis(i * WINDOW), i as f64 + 1.0);
			println!(
				"[win-probe] fed window_start={} results={} kinds={:?}",
				i * WINDOW,
				out.len(),
				out.iter().map(|r| (r.span.start, r.kind)).collect::<Vec<_>>()
			);
			if !out.is_empty() {
				emitted_windows.push(i * WINDOW);
			}
		}
		assert_eq!(
			emitted_windows,
			vec![0, WINDOW, 2 * WINDOW, 3 * WINDOW, 4 * WINDOW],
			"each window that received an event must publish; a ladder that stops after the first \
			 window is the production freeze"
		);
	}

	#[test]
	fn meta_survives_while_group_high_water_at_or_after_threshold() {
		// An active group whose high water is at or beyond the threshold must keep its meta: the
		// carry it holds still seeds the next window.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(Some(millis(2 * WINDOW))));
		for i in 0..3u64 {
			feed(&mut engine, &mut store, at_millis(i * WINDOW), i as f64);
		}
		let dropped = engine.expire_meta(&mut store, WINDOW).unwrap();
		assert_eq!(dropped, 0, "high water (2*WINDOW) is not below the threshold (WINDOW)");
		assert_eq!(store.meta_entry_count(), 1, "an active group within the horizon keeps its meta");
		assert!(store.accumulator_count() > 0, "live windows within retention keep their accumulators");
	}

	#[test]
	fn meta_reclaimed_when_group_stale_past_threshold() {
		// A carry group whose high water falls below the threshold is dead, and the sweep reclaims
		// its meta and sealed carry; otherwise `persist_meta` leaks one key per group forever.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(Some(millis(2 * WINDOW))));
		for i in 0..3u64 {
			feed(&mut engine, &mut store, at_millis(i * WINDOW), i as f64);
		}
		assert_eq!(store.meta_entry_count(), 1);

		let dropped = engine.expire_meta(&mut store, order(100 * WINDOW)).unwrap();
		assert_eq!(dropped, 1, "the quiet group's high water is far below the threshold");
		assert_eq!(store.meta_entry_count(), 0, "a dead carry group must not leak its meta");
	}

	#[test]
	fn without_retention_every_window_accumulator_is_retained() {
		// The contrast that proves the bound above comes from sealing and not some other cap: with
		// no retention configured the engine keeps every window's accumulator forever.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(None));
		for i in 0..60u64 {
			feed(&mut engine, &mut store, at_millis(i * WINDOW), i as f64);
		}
		assert_eq!(
			store.accumulator_count(),
			60,
			"with no retention the carry engine retains every window's accumulator row"
		);
	}

	#[test]
	fn terminal_remove_after_restart_uses_persisted_last_output() {
		// A carry window's withdrawn value cannot be recomputed from its own surviving state: an
		// emptied accumulator finalizes to nothing, and the output also depended on the value
		// carried in from earlier windows. `last_output` must therefore be durable, not ephemeral.
		let mut store = CountingStore::default();

		let mut engine = Engine::new(carry_config(None));
		feed(&mut engine, &mut store, at_millis(0), 5.0);

		let mut engine = Engine::new(carry_config(None));
		let span = WindowSpan::for_coord(at_millis(0), millis(WINDOW));
		let mut buckets: TumblingBuckets<String, DateTime, (u64, f64)> = BTreeMap::new();
		buckets.insert(("BTC".to_string(), span), vec![AccumulatorEvent::Remove((0, 5.0))]);
		let withdrawn: Vec<WindowResult<String, DateTime, f64>> = engine
			.apply(
				&mut store,
				buckets,
				|g: &String, w: DateTime| EncodedKey::builder().str(g).u64(w.to_order()).build(),
				RetainedAccumulator::<u64, f64>::default,
				|_g: &String, _s: WindowSpan<DateTime>, v: &BTreeMap<u64, f64>, _p: Option<&f64>| {
					(!v.is_empty()).then(|| v.values().sum::<f64>())
				},
				|v: &BTreeMap<u64, f64>, _p: Option<&f64>| v.last_key_value().map(|(_, val)| *val),
			)
			.expect("apply");

		assert_eq!(withdrawn.len(), 1, "emptying the window emits exactly one terminal diff");
		assert!(
			matches!(withdrawn[0].kind, EmitKind::Remove),
			"the window emptied under retraction, so the last published row must be withdrawn"
		);
		assert_eq!(
			withdrawn[0].value, 5.0,
			"the withdrawn value is the persisted last_output, recovered across the restart"
		);
	}

	#[test]
	fn last_output_survives_lru_eviction() {
		// The other way the persisted state is read back is LRU eviction, with no restart: the
		// accumulator cache holds 8 windows, so tracking more evicts the oldest and the next access
		// re-reads it from the store.
		let mut store = CountingStore::default();
		let mut engine = Engine::new(carry_config(None));

		let mut published_g00: Vec<WindowResult<String, DateTime, f64>> = Vec::new();
		for i in 0..11u64 {
			let group = format!("G{i:02}");
			let out = feed_group(&mut engine, &mut store, &group, at_millis(0), (i + 1) as f64);
			if i == 0 {
				published_g00 = out;
			}
		}
		assert_eq!(published_g00.len(), 1);
		assert!(matches!(published_g00[0].kind, EmitKind::Insert));
		assert_eq!(published_g00[0].value, 1.0);

		// G00's window was pushed out of the 8-slot accumulator cache by the later groups, so the
		// engine must re-read its accumulator from the store to apply this retraction.
		let span = WindowSpan::for_coord(at_millis(0), millis(WINDOW));
		let mut buckets: TumblingBuckets<String, DateTime, (u64, f64)> = BTreeMap::new();
		buckets.insert(("G00".to_string(), span), vec![AccumulatorEvent::Remove((0, 1.0))]);
		let withdrawn: Vec<WindowResult<String, DateTime, f64>> = engine
			.apply(
				&mut store,
				buckets,
				|g: &String, w: DateTime| EncodedKey::builder().str(g).u64(w.to_order()).build(),
				RetainedAccumulator::<u64, f64>::default,
				|_g: &String, _s: WindowSpan<DateTime>, v: &BTreeMap<u64, f64>, _p: Option<&f64>| {
					(!v.is_empty()).then(|| v.values().sum::<f64>())
				},
				|v: &BTreeMap<u64, f64>, _p: Option<&f64>| v.last_key_value().map(|(_, val)| *val),
			)
			.expect("apply");

		assert_eq!(withdrawn.len(), 1, "emptying the evicted window emits exactly one terminal diff");
		assert!(
			matches!(withdrawn[0].kind, EmitKind::Remove),
			"the evicted window emptied under retraction, so the last published row must be withdrawn"
		);
		assert_eq!(
			withdrawn[0].value, 1.0,
			"the withdrawn value is the persisted last_output for G00, recovered after eviction"
		);
		assert_eq!(
			withdrawn[0].row_number, published_g00[0].row_number,
			"the withdrawal targets the same row that was published for G00"
		);
	}

	#[test]
	fn carry_meta_projects_its_high_water_independently_of_its_window_map() {
		// The meta sweep reclaims on this projection alone, so window entries must never skew it.
		let mut meta: CarryMeta<DateTime, i64, i64> = CarryMeta::default();
		let empty_bytes = meta.encode_state(DateTime::EPOCH).unwrap();
		assert_eq!(
			decode::<CarryMeta<DateTime, i64, i64>>(&empty_bytes).unwrap().high_water_order(),
			None,
			"a default CarryMeta has no high water"
		);

		meta.high_water = Some(at_millis(99));
		meta.windows.insert(
			at_millis(10),
			WindowEntry {
				span: WindowSpan::new(at_millis(10), at_millis(20)),
				carry_out: Some(7i64),
				last_output: Some(3i64),
			},
		);
		let bytes = meta.encode_state(DateTime::EPOCH).unwrap();
		let projected = decode::<CarryMeta<DateTime, i64, i64>>(&bytes).unwrap().high_water_order();
		assert_eq!(projected, Some(order(99)), "the populated window map must not disturb the high water");
	}
}