Skip to content

Commit 80bcabf

Browse files
Saga4Saga4
committed
fix(slack): retain incremental replay evidence
1 parent 01241a7 commit 80bcabf

5 files changed

Lines changed: 456 additions & 63 deletions

File tree

‎Cargo.lock‎

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎crates/locality-slack/Cargo.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ reqwest = { version = "0.13", default-features = false, features = ["blocking",
1919
rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] }
2020
serde = { version = "1.0", features = ["derive"] }
2121
serde_json = "1.0"
22+
sha2 = "0.10"
2223
tokio = { version = "1", features = ["macros", "rt", "sync", "time"] }
2324

2425
[dev-dependencies]

‎crates/locality-slack/src/portable/hosted/checkpoint.rs‎

Lines changed: 139 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ pub const HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V4: u16 = 4;
2626
pub const MAX_HOSTED_SLACK_CURSOR_BYTES_V1: usize = 1024;
2727
pub const MAX_HOSTED_SLACK_APPLIED_PAGES_V1: usize = 256;
2828
pub const MAX_HOSTED_SLACK_REPLAY_BYTES_V1: usize = 512 * 1024;
29+
pub const MAX_HOSTED_SLACK_COMPACT_REPLAY_BYTES_V4: usize = 1024 * 1024;
2930
pub const MAX_HOSTED_SLACK_CHECKPOINT_BYTES_V1: usize = 2 * 1024 * 1024;
3031

3132
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -103,6 +104,25 @@ pub struct HostedSlackAppliedPageV1 {
103104
pub canonical_page_json: String,
104105
}
105106

107+
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
108+
#[serde(deny_unknown_fields)]
109+
pub struct HostedSlackAppliedPageFingerprintV4 {
110+
pub phase: HostedSlackPollPhaseV1,
111+
pub root_message_id: Option<String>,
112+
pub request_cursor: Option<String>,
113+
pub next_cursor: Option<String>,
114+
pub canonical_page_sha256: String,
115+
}
116+
117+
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
118+
#[serde(deny_unknown_fields)]
119+
pub struct HostedSlackObservedMessageFingerprintV4 {
120+
pub message_id: String,
121+
pub thread_root_message_id: Option<String>,
122+
pub canonical_message_sha256: String,
123+
pub history_reply_count: Option<u32>,
124+
}
125+
106126
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
107127
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
108128
pub enum HostedSlackPollEvidenceV1 {
@@ -120,6 +140,12 @@ pub enum HostedSlackPollEvidenceV2 {
120140
AppliedPage {
121141
page: HostedSlackAppliedPageV1,
122142
},
143+
AppliedPageFingerprint {
144+
page: HostedSlackAppliedPageFingerprintV4,
145+
},
146+
ObservedMessageFingerprint {
147+
message: HostedSlackObservedMessageFingerprintV4,
148+
},
123149
BeginCatchUp {
124150
poll_cut_at: String,
125151
},
@@ -538,6 +564,12 @@ impl HostedSlackPollCheckpointV1 {
538564
HostedSlackPollEvidenceV2::AppliedPage { page } => {
539565
super::poll::replay_applied_page_evidence(&mut rebuilt, &page)?;
540566
}
567+
HostedSlackPollEvidenceV2::AppliedPageFingerprint { .. }
568+
| HostedSlackPollEvidenceV2::ObservedMessageFingerprint { .. } => {
569+
return Err(HostedSlackPollError::IncompleteCandidate(
570+
"compact incremental replay evidence",
571+
));
572+
}
541573
HostedSlackPollEvidenceV2::BeginCatchUp { poll_cut_at } => {
542574
rebuilt.begin_catch_up(poll_cut_at)?;
543575
}
@@ -1136,14 +1168,19 @@ fn validate_candidate(
11361168
}
11371169

11381170
fn validate_evidence(checkpoint: &HostedSlackPollCheckpointV1) -> Result<(), HostedSlackPollError> {
1139-
if checkpoint.evidence.len() > MAX_HOSTED_SLACK_APPLIED_PAGES_V1 + 2 {
1171+
if checkpoint.evidence.len()
1172+
> MAX_HOSTED_SLACK_COLLECTION_ENTRIES * 2 + MAX_HOSTED_SLACK_APPLIED_PAGES_V1 + 2
1173+
{
11401174
return Err(HostedSlackPollError::CollectionTooLarge("evidence"));
11411175
}
11421176
let mut keys = BTreeSet::new();
1177+
let mut observed_message_keys = BTreeSet::new();
11431178
let mut replay_bytes = 0usize;
11441179
let mut transitions = 0usize;
11451180
let mut baselines = 0usize;
11461181
let mut applied_pages = 0usize;
1182+
let mut compact_pages = 0usize;
1183+
let mut observed_messages = 0usize;
11471184
for evidence in &checkpoint.evidence {
11481185
match evidence {
11491186
HostedSlackPollEvidenceV2::IncrementalBaseline {
@@ -1193,6 +1230,88 @@ fn validate_evidence(checkpoint: &HostedSlackPollCheckpointV1) -> Result<(), Hos
11931230
}
11941231
replay_bytes = replay_bytes.saturating_add(page.canonical_page_json.len());
11951232
}
1233+
HostedSlackPollEvidenceV2::AppliedPageFingerprint { page } => {
1234+
if checkpoint.checkpoint_format_version
1235+
< HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4
1236+
|| checkpoint.poll_kind != HostedSlackPollKindV2::Incremental
1237+
{
1238+
return Err(HostedSlackPollError::IncompleteCandidate(
1239+
"compact page evidence version",
1240+
));
1241+
}
1242+
compact_pages += 1;
1243+
if compact_pages
1244+
> MAX_HOSTED_SLACK_COLLECTION_ENTRIES + MAX_HOSTED_SLACK_APPLIED_PAGES_V1
1245+
{
1246+
return Err(HostedSlackPollError::CollectionTooLarge(
1247+
"compact page evidence",
1248+
));
1249+
}
1250+
validate_cursor(
1251+
"compact page request_cursor",
1252+
page.request_cursor.as_deref(),
1253+
)?;
1254+
validate_cursor("compact page next_cursor", page.next_cursor.as_deref())?;
1255+
if let Some(root) = &page.root_message_id {
1256+
parse_slack_timestamp("compact page root_message_id", root)?;
1257+
}
1258+
validate_sha256("compact page sha256", &page.canonical_page_sha256)?;
1259+
if !keys.insert((
1260+
page.phase,
1261+
page.root_message_id.as_deref(),
1262+
page.request_cursor.as_deref(),
1263+
)) {
1264+
return Err(HostedSlackPollError::DuplicateValue("applied page key"));
1265+
}
1266+
replay_bytes = replay_bytes
1267+
.saturating_add(page.canonical_page_sha256.len())
1268+
.saturating_add(page.root_message_id.as_ref().map_or(0, String::len))
1269+
.saturating_add(page.request_cursor.as_ref().map_or(0, String::len))
1270+
.saturating_add(page.next_cursor.as_ref().map_or(0, String::len));
1271+
}
1272+
HostedSlackPollEvidenceV2::ObservedMessageFingerprint { message } => {
1273+
if checkpoint.checkpoint_format_version
1274+
< HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4
1275+
|| checkpoint.poll_kind != HostedSlackPollKindV2::Incremental
1276+
{
1277+
return Err(HostedSlackPollError::IncompleteCandidate(
1278+
"observed message evidence version",
1279+
));
1280+
}
1281+
observed_messages += 1;
1282+
if observed_messages > MAX_HOSTED_SLACK_COLLECTION_ENTRIES {
1283+
return Err(HostedSlackPollError::CollectionTooLarge(
1284+
"observed message evidence",
1285+
));
1286+
}
1287+
parse_slack_timestamp("observed message id", &message.message_id)?;
1288+
if !observed_message_keys.insert(message.message_id.as_str()) {
1289+
return Err(HostedSlackPollError::DuplicateValue(
1290+
"observed message evidence",
1291+
));
1292+
}
1293+
if let Some(root) = &message.thread_root_message_id {
1294+
parse_slack_timestamp("observed message root", root)?;
1295+
}
1296+
if message
1297+
.history_reply_count
1298+
.is_some_and(|count| count as usize > MAX_HOSTED_SLACK_THREAD_REPLIES)
1299+
{
1300+
return Err(HostedSlackPollError::CollectionTooLarge(
1301+
"observed history reply count",
1302+
));
1303+
}
1304+
validate_sha256("observed message sha256", &message.canonical_message_sha256)?;
1305+
replay_bytes = replay_bytes
1306+
.saturating_add(message.message_id.len())
1307+
.saturating_add(
1308+
message
1309+
.thread_root_message_id
1310+
.as_ref()
1311+
.map_or(0, String::len),
1312+
)
1313+
.saturating_add(message.canonical_message_sha256.len());
1314+
}
11961315
HostedSlackPollEvidenceV2::BeginCatchUp { poll_cut_at } => {
11971316
parse_canonical_utc_timestamp("evidence.poll_cut_at", poll_cut_at)?;
11981317
transitions += 1;
@@ -1210,16 +1329,33 @@ fn validate_evidence(checkpoint: &HostedSlackPollCheckpointV1) -> Result<(), Hos
12101329
"incremental baseline evidence",
12111330
));
12121331
}
1213-
if replay_bytes > MAX_HOSTED_SLACK_REPLAY_BYTES_V1 {
1332+
let maximum_replay_bytes =
1333+
if checkpoint.checkpoint_format_version >= HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4 {
1334+
MAX_HOSTED_SLACK_COMPACT_REPLAY_BYTES_V4
1335+
} else {
1336+
MAX_HOSTED_SLACK_REPLAY_BYTES_V1
1337+
};
1338+
if replay_bytes > maximum_replay_bytes {
12141339
return Err(HostedSlackPollError::InputTooLarge {
12151340
input: "checkpoint replay evidence",
1216-
maximum_bytes: MAX_HOSTED_SLACK_REPLAY_BYTES_V1,
1341+
maximum_bytes: maximum_replay_bytes,
12171342
actual_bytes: replay_bytes,
12181343
});
12191344
}
12201345
Ok(())
12211346
}
12221347

1348+
fn validate_sha256(field: &'static str, value: &str) -> Result<(), HostedSlackPollError> {
1349+
if value.len() != 64
1350+
|| !value
1351+
.bytes()
1352+
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
1353+
{
1354+
return Err(HostedSlackPollError::IncompleteCandidate(field));
1355+
}
1356+
Ok(())
1357+
}
1358+
12231359
fn ensure_unique<'a>(
12241360
field: &'static str,
12251361
values: impl Iterator<Item = &'a str>,

0 commit comments

Comments
 (0)