From 944b1d75b6d1011301db159cf082bbe57ba15860 Mon Sep 17 00:00:00 2001 From: Derek Date: Tue, 10 Nov 2026 20:54:42 +1100 Subject: [PATCH] fix(tests): read DLQ entries through scalo scalo 2.14.28 writes a dead letter as a DFE event, with the original under `_json` or under `_raw` with an optional `_raw_encoding`. It no longer writes a base64 `payload` key. The integration helpers read `payload` and found no entry. Every full-tier test that checks the file DLQ failed. - The helpers parse each line as `scalo::dlq::DlqEntry`, which restores the original bytes from either field. - The hand-written base64 decoder is gone because scalo's parser decodes `_raw_encoding: base64`. --- tests/integration/grpc_outage.rs | 79 ++++++++++++-------------------- tests/integration/json_depth.rs | 37 +++------------ 2 files changed, 37 insertions(+), 79 deletions(-) diff --git a/tests/integration/grpc_outage.rs b/tests/integration/grpc_outage.rs index f5b22fcf..e3b39c35 100644 --- a/tests/integration/grpc_outage.rs +++ b/tests/integration/grpc_outage.rs @@ -29,6 +29,7 @@ use dfe_loader::config::Config; use dfe_loader::metrics::{Metrics, ServerState}; use dfe_loader::pipeline::Orchestrator; use scalo::ScalingPressure; +use scalo::dlq::DlqEntry; use scalo::metrics::MetricsManager; use scalo::scaling::ScalingPressureConfig; use scalo::transport::grpc::proto::transport_client::TransportClient; @@ -486,68 +487,48 @@ pub(super) fn counter_value(manager: &MetricsManager, name: &str, labels: &str) .sum() } -/// Decode standard-alphabet base64, the encoding the file DLQ gives a payload. -pub(super) fn decode_base64(text: &str) -> Option> { - let mut out = Vec::with_capacity(text.len() * 3 / 4); - let mut acc = 0_u32; - let mut bits = 0_u32; - for c in text.bytes().take_while(|&c| c != b'=') { - let value = match c { - b'A'..=b'Z' => c - b'A', - b'a'..=b'z' => c - b'a' + 26, - b'0'..=b'9' => c - b'0' + 52, - b'+' => 62, - b'/' => 63, - _ => return None, - }; - acc = (acc << 6) | u32::from(value); - bits += 6; - if bits >= 8 { - bits -= 8; - out.push((acc >> bits) as u8); - } - } - Some(out) -} - -/// Every id the file DLQ under `dir` has written, skipping any torn line. -pub(super) fn dead_lettered_ids(dir: &Path) -> BTreeSet { - dead_letter_reasons(dir).into_keys().collect() -} - -/// The reason the file DLQ under `dir` recorded for each id it has written, -/// skipping any torn line. -pub(super) fn dead_letter_reasons(dir: &Path) -> std::collections::BTreeMap { - fn walk(dir: &Path, reasons: &mut std::collections::BTreeMap) { +/// Every entry the file DLQ under `dir` has written, read back through scalo's +/// own entry format, skipping any torn line. +pub(super) fn dead_letters(dir: &Path) -> Vec { + fn walk(dir: &Path, out: &mut Vec) { let Ok(entries) = std::fs::read_dir(dir) else { return; }; for entry in entries.flatten() { let path = entry.path(); if path.is_dir() { - walk(&path, reasons); + walk(&path, out); continue; } let Ok(body) = std::fs::read_to_string(&path) else { continue; }; - for line in body.lines() { - let Ok(entry) = serde_json::from_str::(line) else { - continue; - }; - let id = decode_base64(entry["payload"].as_str().unwrap_or_default()) - .and_then(|payload| serde_json::from_slice::(&payload).ok()) - .and_then(|record| record["id"].as_u64()); - if let Some(id) = id { - let reason = entry["reason"].as_str().unwrap_or_default().to_string(); - reasons.insert(id, reason); - } - } + out.extend( + body.lines() + .filter_map(|line| serde_json::from_str::(line).ok()), + ); } } - let mut reasons = std::collections::BTreeMap::new(); - walk(dir, &mut reasons); - reasons + let mut out = Vec::new(); + walk(dir, &mut out); + out +} + +/// Every id the file DLQ under `dir` has written, skipping any torn line. +pub(super) fn dead_lettered_ids(dir: &Path) -> BTreeSet { + dead_letter_reasons(dir).into_keys().collect() +} + +/// The reason the file DLQ under `dir` recorded for each id it has written, +/// skipping any torn line. +pub(super) fn dead_letter_reasons(dir: &Path) -> std::collections::BTreeMap { + dead_letters(dir) + .into_iter() + .filter_map(|entry| { + let record = serde_json::from_slice::(&entry.payload).ok()?; + Some((record["id"].as_u64()?, entry.reason)) + }) + .collect() } /// Wait until the DLQ holds every id in `want`, or the budget runs out. diff --git a/tests/integration/json_depth.rs b/tests/integration/json_depth.rs index f8a0b687..b68c5ec7 100644 --- a/tests/integration/json_depth.rs +++ b/tests/integration/json_depth.rs @@ -19,8 +19,8 @@ use std::path::Path; use std::time::Duration; use super::grpc_outage::{ - clickhouse_address, decode_base64, free_port, grpc_loader, push_ids, query_client, - start_loader, wait_for_loader, wait_landed, with_file_dlq, + clickhouse_address, dead_letters, free_port, grpc_loader, push_ids, query_client, start_loader, + wait_for_loader, wait_landed, with_file_dlq, }; use crate::common::containers::TestInfrastructure; use crate::common::unique_table_name; @@ -42,34 +42,11 @@ fn nested_object(depth: usize) -> String { /// Every payload the file DLQ under `dir` holds with the depth reason. fn dead_lettered_too_deep(dir: &Path) -> Vec> { - fn walk(dir: &Path, out: &mut Vec>) { - let Ok(entries) = std::fs::read_dir(dir) else { - return; - }; - for entry in entries.flatten() { - let path = entry.path(); - if path.is_dir() { - walk(&path, out); - continue; - } - let Ok(body) = std::fs::read_to_string(&path) else { - continue; - }; - for line in body.lines() { - let Ok(entry) = serde_json::from_str::(line) else { - continue; - }; - if entry["reason"].as_str() == Some(TOO_DEEP) - && let Some(payload) = entry["payload"].as_str().and_then(decode_base64) - { - out.push(payload); - } - } - } - } - let mut out = Vec::new(); - walk(dir, &mut out); - out + dead_letters(dir) + .into_iter() + .filter(|entry| entry.reason == TOO_DEEP) + .map(|entry| entry.payload) + .collect() } #[test]