///|
pub(all) struct RetentionPolicy {
name : String
topic : TopicPattern?
min_timestamp_ms : Int?
max_timestamp_ms : Int?
max_events : Int
max_events_per_topic : Int
compact_fingerprints : Bool
} derive(Eq, @debug.Debug)
///|
pub(all) struct RetentionDecision {
index : Int
sequence : Int
event_id : String
topic : String
timestamp_ms : Int
fingerprint : String
keep : Bool
reason : String
} derive(Eq, @debug.Debug)
///|
pub(all) struct RetentionTopicSummary {
topic : String
kept : Int
dropped : Int
} derive(Eq, @debug.Debug)
///|
pub(all) struct RetentionReport {
policy : String
source : String
scanned : Int
kept : Int
dropped : Int
topic_summaries : Array[RetentionTopicSummary]
decisions : Array[RetentionDecision]
} derive(Eq, @debug.Debug)
///|
struct RetentionCandidate {
index : Int
sequence : Int
event : Envelope
} derive(Eq, @debug.Debug)
///|
struct RetentionCounter {
topic : String
kept : Int
} derive(Eq, @debug.Debug)
///|
pub fn retention_policy(
name? : StringView = "retention",
topic? : StringView = "",
min_timestamp_ms? : Int,
max_timestamp_ms? : Int,
max_events? : Int = 0,
max_events_per_topic? : Int = 0,
compact_fingerprints? : Bool = false,
) -> Result[RetentionPolicy, EventRailError] {
let parsed = if topic == "" {
None
} else {
match topic_pattern(topic) {
Ok(value) => Some(value)
Err(err) => return Err(err)
}
}
Ok({
name: name.to_owned(),
topic: parsed,
min_timestamp_ms,
max_timestamp_ms,
max_events: clamp_non_negative(max_events),
max_events_per_topic: clamp_non_negative(max_events_per_topic),
compact_fingerprints,
})
}
///|
pub fn RetentionPolicy::with_topic(
self : RetentionPolicy,
topic : StringView,
) -> Result[RetentionPolicy, EventRailError] {
match topic_pattern(topic) {
Ok(value) => Ok({ ..self, topic: Some(value) })
Err(err) => Err(err)
}
}
///|
pub fn RetentionPolicy::without_topic(
self : RetentionPolicy,
) -> RetentionPolicy {
{ ..self, topic: None }
}
///|
pub fn RetentionPolicy::with_time_range(
self : RetentionPolicy,
min_timestamp_ms? : Int,
max_timestamp_ms? : Int,
) -> RetentionPolicy {
{ ..self, min_timestamp_ms, max_timestamp_ms }
}
///|
pub fn RetentionPolicy::with_limits(
self : RetentionPolicy,
max_events? : Int = 0,
max_events_per_topic? : Int = 0,
) -> RetentionPolicy {
{
..self,
max_events: clamp_non_negative(max_events),
max_events_per_topic: clamp_non_negative(max_events_per_topic),
}
}
///|
pub fn RetentionPolicy::with_compaction(
self : RetentionPolicy,
compact_fingerprints : Bool,
) -> RetentionPolicy {
{ ..self, compact_fingerprints, }
}
///|
pub fn RetentionPolicy::to_wire(self : RetentionPolicy) -> String {
let topic = match self.topic {
Some(pattern) => pattern.raw
None => "*"
}
let min_value = match self.min_timestamp_ms {
Some(value) => value.to_string()
None => "*"
}
let max_value = match self.max_timestamp_ms {
Some(value) => value.to_string()
None => "*"
}
"retention=\{escape_wire_text(self.name)};topic=\{escape_wire_text(topic)};min=\{min_value};max=\{max_value};max_events=\{self.max_events};max_per_topic=\{self.max_events_per_topic};compact=\{self.compact_fingerprints}"
}
///|
pub fn RetentionPolicy::bounded(self : RetentionPolicy) -> Bool {
self.topic is Some(_) ||
self.min_timestamp_ms is Some(_) ||
self.max_timestamp_ms is Some(_) ||
self.max_events > 0 ||
self.max_events_per_topic > 0 ||
self.compact_fingerprints
}
///|
pub fn EventBatch::retention_plan(
self : EventBatch,
policy : RetentionPolicy,
) -> Result[RetentionReport, EventRailError] {
let candidates : Array[RetentionCandidate] = []
for index, event in self.events {
candidates.push({ index, sequence: index, event })
}
build_retention_report("batch:\{self.name}", candidates, policy)
}
///|
pub fn EventTape::retention_plan(
self : EventTape,
policy : RetentionPolicy,
) -> Result[RetentionReport, EventRailError] {
let candidates : Array[RetentionCandidate] = []
for index, entry in self.entries {
candidates.push({ index, sequence: entry.sequence, event: entry.event })
}
build_retention_report("event-tape", candidates, policy)
}
///|
pub fn EventBatch::apply_retention(
self : EventBatch,
policy : RetentionPolicy,
) -> Result[(EventBatch, RetentionReport), EventRailError] {
match self.retention_plan(policy) {
Err(err) => Err(err)
Ok(report) => {
let kept = report.kept_indices()
let events : Array[Envelope] = []
for index, event in self.events {
if kept.contains(index) {
events.push(event)
}
}
Ok(({ ..self, events, }, report))
}
}
}
///|
pub fn EventTape::apply_retention(
self : EventTape,
policy : RetentionPolicy,
) -> Result[(EventTape, RetentionReport), EventRailError] {
match self.retention_plan(policy) {
Err(err) => Err(err)
Ok(report) => {
let kept = report.kept_sequences()
let entries : Array[TapeEntry] = []
for entry in self.entries {
if kept.contains(entry.sequence) {
entries.push({ ..entry, sequence: entries.length() })
}
}
Ok(({ entries, next_sequence: entries.length() }, report))
}
}
}
///|
pub fn RetentionReport::ok(self : RetentionReport) -> Bool {
self.kept > 0 || self.scanned == 0
}
///|
pub fn RetentionReport::summary(self : RetentionReport) -> String {
"retention=\{escape_wire_text(self.policy)} source=\{escape_wire_text(self.source)} scanned=\{self.scanned} kept=\{self.kept} dropped=\{self.dropped} topics=\{self.topic_summaries.length()}"
}
///|
pub fn RetentionReport::kept_indices(self : RetentionReport) -> Array[Int] {
self.decisions
.filter(decision => decision.keep)
.map(decision => decision.index)
}
///|
pub fn RetentionReport::dropped_indices(self : RetentionReport) -> Array[Int] {
self.decisions
.filter(decision => !decision.keep)
.map(decision => decision.index)
}
///|
pub fn RetentionReport::kept_sequences(self : RetentionReport) -> Array[Int] {
self.decisions
.filter(decision => decision.keep)
.map(decision => decision.sequence)
}
///|
pub fn RetentionReport::dropped_sequences(self : RetentionReport) -> Array[Int] {
self.decisions
.filter(decision => !decision.keep)
.map(decision => decision.sequence)
}
///|
pub fn RetentionReport::kept_event_ids(self : RetentionReport) -> Array[String] {
self.decisions
.filter(decision => decision.keep)
.map(decision => decision.event_id)
}
///|
pub fn RetentionReport::dropped_event_ids(
self : RetentionReport,
) -> Array[String] {
self.decisions
.filter(decision => !decision.keep)
.map(decision => decision.event_id)
}
///|
pub fn RetentionReport::decision_lines(self : RetentionReport) -> Array[String] {
self.decisions.map(decision => decision.to_wire())
}
///|
pub fn RetentionReport::keep_lines(self : RetentionReport) -> Array[String] {
self.decisions
.filter(decision => decision.keep)
.map(decision => decision.to_wire())
}
///|
pub fn RetentionReport::drop_lines(self : RetentionReport) -> Array[String] {
self.decisions
.filter(decision => !decision.keep)
.map(decision => decision.to_wire())
}
///|
pub fn RetentionReport::topic_lines(self : RetentionReport) -> Array[String] {
self.topic_summaries.map(row => row.to_wire())
}
///|
pub fn RetentionReport::manifest_lines(self : RetentionReport) -> Array[String] {
let lines : Array[String] = []
lines.push(self.summary())
for topic in self.topic_summaries {
lines.push(topic.to_wire())
}
for decision in self.decisions {
lines.push(decision.to_wire())
}
lines
}
///|
pub fn RetentionReport::manifest(self : RetentionReport) -> String {
self.manifest_lines().join("\n")
}
///|
pub fn RetentionDecision::to_wire(self : RetentionDecision) -> String {
"idx=\{self.index};seq=\{self.sequence};event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};timestamp=\{self.timestamp_ms};keep=\{self.keep};reason=\{escape_wire_text(self.reason)}"
}
///|
pub fn RetentionTopicSummary::to_wire(self : RetentionTopicSummary) -> String {
"topic=\{escape_wire_text(self.topic)};kept=\{self.kept};dropped=\{self.dropped}"
}
///|
fn build_retention_report(
source : StringView,
candidates : Array[RetentionCandidate],
policy : RetentionPolicy,
) -> Result[RetentionReport, EventRailError] {
let decisions : Array[RetentionDecision] = []
let counters : Array[RetentionCounter] = []
let seen_fingerprints : Array[String] = []
let mut kept = 0
let mut dropped = 0
for offset in 0.. return Err(err)
Ok((true, reason, fingerprint)) => {
kept += 1
increment_retention_counter(counters, candidate.event.topic)
if policy.compact_fingerprints &&
!seen_fingerprints.contains(fingerprint) {
seen_fingerprints.push(fingerprint)
}
decisions.push(retention_decision(candidate, true, reason, fingerprint))
}
Ok((false, reason, fingerprint)) => {
dropped += 1
decisions.push(
retention_decision(candidate, false, reason, fingerprint),
)
}
}
}
decisions.sort_by(compare_retention_decision)
Ok({
policy: policy.name,
source: source.to_owned(),
scanned: candidates.length(),
kept,
dropped,
topic_summaries: summarize_retention_topics(decisions),
decisions,
})
}
///|
fn decide_retention(
candidate : RetentionCandidate,
policy : RetentionPolicy,
kept_total : Int,
counters : Array[RetentionCounter],
seen_fingerprints : Array[String],
) -> Result[(Bool, String, String), EventRailError] {
match topic_segments(candidate.event.topic) {
Err(err) => return Err(err)
Ok(_) => ()
}
match policy.topic {
Some(pattern) =>
match pattern.matches_topic(candidate.event.topic) {
Err(err) => return Err(err)
Ok(false) =>
return Ok((false, "topic mismatch", candidate.event.fingerprint()))
Ok(true) => ()
}
None => ()
}
match policy.min_timestamp_ms {
Some(minimum) if candidate.event.timestamp_ms < minimum =>
return Ok(
(false, "timestamp before minimum", candidate.event.fingerprint()),
)
_ => ()
}
match policy.max_timestamp_ms {
Some(maximum) if candidate.event.timestamp_ms > maximum =>
return Ok(
(false, "timestamp after maximum", candidate.event.fingerprint()),
)
_ => ()
}
let fingerprint = candidate.event.fingerprint()
if policy.compact_fingerprints && seen_fingerprints.contains(fingerprint) {
return Ok((false, "duplicate fingerprint", fingerprint))
}
if policy.max_events > 0 && kept_total >= policy.max_events {
return Ok((false, "total limit reached", fingerprint))
}
if policy.max_events_per_topic > 0 &&
retained_topic_count(counters, candidate.event.topic) >=
policy.max_events_per_topic {
return Ok((false, "topic limit reached", fingerprint))
}
Ok((true, "retained", fingerprint))
}
///|
fn retention_decision(
candidate : RetentionCandidate,
keep : Bool,
reason : String,
fingerprint : String,
) -> RetentionDecision {
{
index: candidate.index,
sequence: candidate.sequence,
event_id: candidate.event.id,
topic: candidate.event.topic,
timestamp_ms: candidate.event.timestamp_ms,
fingerprint,
keep,
reason,
}
}
///|
fn summarize_retention_topics(
decisions : Array[RetentionDecision],
) -> Array[RetentionTopicSummary] {
let rows : Array[RetentionTopicSummary] = []
for decision in decisions {
record_retention_topic(rows, decision.topic, decision.keep)
}
rows.sort_by(compare_retention_topic_summary)
rows
}
///|
fn retained_topic_count(
counters : Array[RetentionCounter],
topic : StringView,
) -> Int {
let wanted = topic.to_owned()
for counter in counters {
if counter.topic == wanted {
return counter.kept
}
}
0
}
///|
fn increment_retention_counter(
counters : Array[RetentionCounter],
topic : String,
) -> Unit {
for index, counter in counters {
if counter.topic == topic {
counters[index] = { ..counter, kept: counter.kept + 1 }
return
}
}
counters.push({ topic, kept: 1 })
}
///|
fn record_retention_topic(
rows : Array[RetentionTopicSummary],
topic : String,
keep : Bool,
) -> Unit {
for index, row in rows {
if row.topic == topic {
rows[index] = if keep {
{ ..row, kept: row.kept + 1 }
} else {
{ ..row, dropped: row.dropped + 1 }
}
return
}
}
rows.push(
if keep {
{ topic, kept: 1, dropped: 0 }
} else {
{ topic, kept: 0, dropped: 1 }
},
)
}
///|
fn compare_retention_decision(
left : RetentionDecision,
right : RetentionDecision,
) -> Int {
if left.index != right.index {
left.index - right.index
} else {
left.sequence - right.sequence
}
}
///|
fn compare_retention_topic_summary(
left : RetentionTopicSummary,
right : RetentionTopicSummary,
) -> Int {
left.topic.compare(right.topic)
}
///|
fn clamp_non_negative(value : Int) -> Int {
if value < 0 {
0
} else {
value
}
}