Skip to content

Commit 01241a7

Browse files
Saga4Saga4
committed
fix(slack): resume bounded incremental root sweeps
1 parent 8f4831a commit 01241a7

3 files changed

Lines changed: 183 additions & 23 deletions

File tree

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

Lines changed: 28 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@ pub const HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V2: u16 = 2;
2121
pub const HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V2: u16 = 2;
2222
pub const HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3: u16 = 3;
2323
pub const HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V3: u16 = 3;
24+
pub const HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4: u16 = 4;
25+
pub const HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V4: u16 = 4;
2426
pub const MAX_HOSTED_SLACK_CURSOR_BYTES_V1: usize = 1024;
2527
pub const MAX_HOSTED_SLACK_APPLIED_PAGES_V1: usize = 256;
2628
pub const MAX_HOSTED_SLACK_REPLAY_BYTES_V1: usize = 512 * 1024;
@@ -380,8 +382,8 @@ impl HostedSlackPollCheckpointV1 {
380382
backfill_cut_at.clone(),
381383
poll_overlap_watermark,
382384
);
383-
checkpoint.checkpoint_format_version = HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3;
384-
checkpoint.minimum_reader_version = HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V3;
385+
checkpoint.checkpoint_format_version = HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4;
386+
checkpoint.minimum_reader_version = HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V4;
385387
checkpoint.install_incremental_baseline(
386388
applied.candidate.clone(),
387389
applied.completed_roots.clone(),
@@ -436,7 +438,11 @@ impl HostedSlackPollCheckpointV1 {
436438
pub fn validate(&self) -> Result<(), HostedSlackPollError> {
437439
self.validate_internal()?;
438440
if (self.poll_kind == HostedSlackPollKindV2::Incremental)
439-
!= (self.checkpoint_format_version == HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3)
441+
!= matches!(
442+
self.checkpoint_format_version,
443+
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3
444+
| HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4
445+
)
440446
{
441447
return Err(HostedSlackPollError::UnsupportedVersion {
442448
format_version: self.checkpoint_format_version,
@@ -457,8 +463,8 @@ impl HostedSlackPollCheckpointV1 {
457463
validate_checkpoint_versions_for_reader(
458464
self.checkpoint_format_version,
459465
self.minimum_reader_version,
460-
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3,
461-
HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V3,
466+
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4,
467+
HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V4,
462468
)?;
463469
let selector = self.selector();
464470
selector
@@ -558,8 +564,8 @@ impl HostedSlackPollCheckpointV1 {
558564
));
559565
}
560566
let mut next = self.clone();
561-
next.checkpoint_format_version = HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3;
562-
next.minimum_reader_version = HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V3;
567+
next.checkpoint_format_version = self.checkpoint_format_version;
568+
next.minimum_reader_version = self.minimum_reader_version;
563569
next.candidate = candidate.clone();
564570
next.completed_roots = completed_roots.clone();
565571
next.latest_observed_message_timestamp = latest_observed_message_timestamp.clone();
@@ -631,6 +637,17 @@ pub fn decode_hosted_slack_poll_checkpoint_v2(
631637
)
632638
}
633639

640+
/// Reads the resumable V4 incremental checkpoint while older readers reject it.
641+
pub fn decode_hosted_slack_poll_checkpoint_v3(
642+
bytes: &[u8],
643+
) -> Result<HostedSlackPollCheckpointV1, HostedSlackPollError> {
644+
decode_hosted_slack_poll_checkpoint_for_reader(
645+
bytes,
646+
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4,
647+
HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V4,
648+
)
649+
}
650+
634651
fn decode_hosted_slack_poll_checkpoint_for_reader(
635652
bytes: &[u8],
636653
supported_format_version: u16,
@@ -799,6 +816,9 @@ pub(crate) fn validate_checkpoint_versions_for_reader(
799816
) | (
800817
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3,
801818
HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V3
819+
) | (
820+
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4,
821+
HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V4
802822
)
803823
) {
804824
return Err(HostedSlackPollError::UnsupportedVersion {
@@ -1138,7 +1158,7 @@ fn validate_evidence(checkpoint: &HostedSlackPollCheckpointV1) -> Result<(), Hos
11381158
}
11391159
if *applied_checkpoint_format_version == 0
11401160
|| *applied_checkpoint_format_version
1141-
> HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V3
1161+
> HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4
11421162
|| applied_poll_cut_at != &checkpoint.backfill_cut_at
11431163
{
11441164
return Err(HostedSlackPollError::IncompleteCandidate(

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

Lines changed: 40 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,10 @@ use locality_protocol::{
77
use serde::{Deserialize, Serialize};
88

99
use super::checkpoint::{
10-
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V2, HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V2,
11-
HostedSlackAppliedPageV1, HostedSlackCompletedRootV1, HostedSlackPollCheckpointV1,
12-
HostedSlackPollError, HostedSlackPollEvidenceV2, HostedSlackPollKindV1, HostedSlackPollKindV2,
10+
HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V2, HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4,
11+
HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V2, HostedSlackAppliedPageV1,
12+
HostedSlackCompletedRootV1, HostedSlackPollCheckpointV1, HostedSlackPollError,
13+
HostedSlackPollEvidenceV2, HostedSlackPollKindV1, HostedSlackPollKindV2,
1314
HostedSlackPollPhaseV1, HostedSlackRootExpectationV1, MAX_HOSTED_SLACK_APPLIED_PAGES_V1,
1415
compare_slack_timestamps, parse_canonical_utc_timestamp, parse_slack_timestamp,
1516
validate_cursor, validate_page_scope_id,
@@ -760,8 +761,11 @@ impl HostedSlackPollCheckpointV1 {
760761
next.checkpoint_format_version = HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V2;
761762
next.minimum_reader_version = HOSTED_SLACK_POLL_MINIMUM_READER_VERSION_V2;
762763
}
764+
let compact_incremental =
765+
next.checkpoint_format_version >= HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4;
763766
let expectation_is_from_this_catch_up = page.phase
764767
!= HostedSlackPollPhaseV1::CatchUpReplies
768+
|| compact_incremental
765769
|| catch_up_history_contains_root(&next, &page.root_message_id)?
766770
|| !first_page;
767771
if deleted_root_reconciliation {
@@ -772,6 +776,14 @@ impl HostedSlackPollCheckpointV1 {
772776
expected_reply_count: 0,
773777
},
774778
);
779+
} else if compact_incremental && first_page {
780+
upsert_root_expectation(
781+
&mut next.candidate.root_expectations,
782+
HostedSlackRootExpectationV1 {
783+
root_message_id: page.root_message_id.clone(),
784+
expected_reply_count: page.root_reply_count,
785+
},
786+
);
775787
} else if expectation_is_from_this_catch_up {
776788
let expected = expected_reply_count(&next, &page.root_message_id)?;
777789
if expected != page.root_reply_count {
@@ -851,6 +863,9 @@ impl HostedSlackPollCheckpointV1 {
851863
page.next_cursor.clone(),
852864
canonical_page_json,
853865
);
866+
if page.next_cursor.is_none() && compact_incremental {
867+
retain_incremental_cursor_evidence(&mut next);
868+
}
854869
rebuild_candidate_reference_metadata(&mut next)?;
855870
if next.phase == HostedSlackPollPhaseV1::CompleteCandidate {
856871
build_snapshot(&next)?;
@@ -1409,16 +1424,22 @@ fn prepare_reply_phase(
14091424
pending.push(root);
14101425
}
14111426
}
1412-
let applied_pages = checkpoint
1413-
.evidence
1414-
.iter()
1415-
.filter(|evidence| matches!(evidence, HostedSlackPollEvidenceV2::AppliedPage { .. }))
1416-
.count();
1417-
let minimum_required_pages = applied_pages.saturating_add(pending.len());
1418-
if minimum_required_pages > MAX_HOSTED_SLACK_APPLIED_PAGES_V1 {
1419-
return Err(HostedSlackPollError::CollectionTooLarge(
1420-
"catch-up root sweep",
1421-
));
1427+
if checkpoint.checkpoint_format_version < HOSTED_SLACK_POLL_CHECKPOINT_FORMAT_VERSION_V4 {
1428+
let applied_pages = checkpoint
1429+
.evidence
1430+
.iter()
1431+
.filter(|evidence| {
1432+
matches!(evidence, HostedSlackPollEvidenceV2::AppliedPage { .. })
1433+
})
1434+
.count();
1435+
let minimum_required_pages = applied_pages.saturating_add(pending.len());
1436+
if minimum_required_pages > MAX_HOSTED_SLACK_APPLIED_PAGES_V1 {
1437+
return Err(HostedSlackPollError::CollectionTooLarge(
1438+
"catch-up root sweep",
1439+
));
1440+
}
1441+
} else {
1442+
retain_incremental_cursor_evidence(checkpoint);
14221443
}
14231444
pending
14241445
};
@@ -1445,6 +1466,12 @@ fn prepare_reply_phase(
14451466
Ok(())
14461467
}
14471468

1469+
fn retain_incremental_cursor_evidence(checkpoint: &mut HostedSlackPollCheckpointV1) {
1470+
checkpoint
1471+
.evidence
1472+
.retain(|evidence| !matches!(evidence, HostedSlackPollEvidenceV2::AppliedPage { .. }));
1473+
}
1474+
14481475
fn complete_current_root(
14491476
checkpoint: &mut HostedSlackPollCheckpointV1,
14501477
reply_phase: HostedSlackPollPhaseV1,

‎crates/locality-slack/tests/hosted_polling.rs‎

Lines changed: 115 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,8 @@ use locality_slack::portable::hosted::{
99
MAX_HOSTED_SLACK_CURSOR_BYTES_V1, MAX_HOSTED_SLACK_POLL_PAGE_BYTES_V1,
1010
MAX_HOSTED_SLACK_POLL_PAGE_MESSAGES_V1, RawHostedSlackMessage, RawHostedSlackNativeSnapshot,
1111
decode_hosted_slack_history_page_v1, decode_hosted_slack_poll_checkpoint_v1,
12-
decode_hosted_slack_poll_checkpoint_v2, decode_hosted_slack_replies_page_v1,
12+
decode_hosted_slack_poll_checkpoint_v2, decode_hosted_slack_poll_checkpoint_v3,
13+
decode_hosted_slack_replies_page_v1,
1314
};
1415
use serde::Serialize;
1516

@@ -1184,9 +1185,10 @@ fn incremental_poll_starts_from_applied_candidate_and_reconciles_late_old_root_r
11841185
"the applied candidate must not be duplicated into replay evidence"
11851186
);
11861187
assert_eq!(
1187-
decode_hosted_slack_poll_checkpoint_v2(&encoded).unwrap(),
1188+
decode_hosted_slack_poll_checkpoint_v3(&encoded).unwrap(),
11881189
incremental
11891190
);
1191+
assert!(decode_hosted_slack_poll_checkpoint_v2(&encoded).is_err());
11901192
assert!(decode_hosted_slack_poll_checkpoint_v1(&encoded).is_err());
11911193
let mut wrong_incremental_version = encoded_value.clone();
11921194
wrong_incremental_version["checkpoint_format_version"] = 2.into();
@@ -1247,6 +1249,117 @@ fn incremental_poll_starts_from_applied_candidate_and_reconciles_late_old_root_r
12471249
);
12481250
}
12491251

1252+
#[test]
1253+
fn incremental_old_root_sweep_resumes_beyond_evidence_page_bound() {
1254+
let mut applied = checkpoint();
1255+
let mut empty_history = history_page();
1256+
empty_history.next_cursor = None;
1257+
empty_history.messages.clear();
1258+
empty_history.users.clear();
1259+
empty_history.files.clear();
1260+
applied.apply_history_page(&empty_history).unwrap();
1261+
applied
1262+
.begin_catch_up("2026-06-02T00:00:00Z".to_string())
1263+
.unwrap();
1264+
1265+
let template = history_page().messages[0].clone();
1266+
let roots = (0..300)
1267+
.map(|offset| {
1268+
let mut root = template.clone();
1269+
root.message.ts = format!("1780272000.{offset:06}");
1270+
root.message.thread_ts = None;
1271+
root.message.user_id = None;
1272+
root.message.file_ids.clear();
1273+
root.reply_count = 0;
1274+
root
1275+
})
1276+
.collect::<Vec<_>>();
1277+
let mut seed = catch_up_page();
1278+
seed.messages.clone_from(&roots);
1279+
applied.apply_history_page(&seed).unwrap();
1280+
assert!(applied.completed_output().is_ok());
1281+
1282+
let mut sweep = HostedSlackPollCheckpointV1::incremental_from_applied(
1283+
&applied,
1284+
raw_snapshot().channel,
1285+
"2026-06-01T23:55:00Z".to_string(),
1286+
)
1287+
.unwrap();
1288+
sweep
1289+
.begin_catch_up("2026-06-02T00:05:00Z".to_string())
1290+
.unwrap();
1291+
let mut empty_incremental: HostedSlackHistoryPageV2 = catch_up_page().into();
1292+
empty_incremental.page_format_version = HOSTED_SLACK_POLL_PAGE_FORMAT_VERSION_V3;
1293+
empty_incremental.minimum_reader_version = HOSTED_SLACK_POLL_PAGE_MINIMUM_READER_VERSION_V3;
1294+
empty_incremental.poll_kind = HostedSlackPollKindV2::Incremental;
1295+
empty_incremental.backfill_cut_at = "2026-06-02T00:00:00Z".to_string();
1296+
empty_incremental.poll_overlap_watermark = "2026-06-01T23:55:00Z".to_string();
1297+
empty_incremental.poll_cut_at = Some("2026-06-02T00:05:00Z".to_string());
1298+
empty_incremental.observed_at = "2026-06-02T00:05:01Z".to_string();
1299+
sweep.apply_history_page_v2(&empty_incremental).unwrap();
1300+
1301+
for (offset, root) in roots.iter().enumerate() {
1302+
assert!(
1303+
sweep.completed_output().is_err(),
1304+
"no false Fresh at {offset}"
1305+
);
1306+
let mut page: HostedSlackRepliesPageV2 = catch_up_replies_page(replies_page()).into();
1307+
page.page_format_version = HOSTED_SLACK_POLL_PAGE_FORMAT_VERSION_V3;
1308+
page.minimum_reader_version = HOSTED_SLACK_POLL_PAGE_MINIMUM_READER_VERSION_V3;
1309+
page.poll_kind = HostedSlackPollKindV2::Incremental;
1310+
page.backfill_cut_at = "2026-06-02T00:00:00Z".to_string();
1311+
page.poll_overlap_watermark = "2026-06-01T23:55:00Z".to_string();
1312+
page.poll_cut_at = Some("2026-06-02T00:05:00Z".to_string());
1313+
page.request_cursor = None;
1314+
page.next_cursor = None;
1315+
page.root_message_id.clone_from(&root.message.ts);
1316+
page.root_reply_count = u32::from(offset == 298);
1317+
page.observed_at = "2026-06-02T00:05:02Z".to_string();
1318+
page.messages = vec![root.message.clone()];
1319+
page.users.clear();
1320+
page.files.clear();
1321+
if offset == 298 {
1322+
let mut late = root.message.clone();
1323+
late.ts = "1780272200.000001".to_string();
1324+
late.thread_ts = Some(root.message.ts.clone());
1325+
late.text = "late near sweep end".to_string();
1326+
page.messages.push(late);
1327+
}
1328+
1329+
if offset == 150 {
1330+
let encoded = serde_json::to_vec(&sweep).unwrap();
1331+
assert!(encoded.len() <= MAX_HOSTED_SLACK_CHECKPOINT_BYTES_V1);
1332+
let mut replay = decode_hosted_slack_poll_checkpoint_v3(&encoded).unwrap();
1333+
sweep.apply_replies_page_v2(&page).unwrap();
1334+
replay.apply_replies_page_v2(&page).unwrap();
1335+
assert_eq!(sweep, replay, "interruption/replay must be exact");
1336+
} else {
1337+
sweep
1338+
.apply_replies_page_v2(&page)
1339+
.unwrap_or_else(|error| panic!("root {offset}: {error:?}"));
1340+
}
1341+
}
1342+
1343+
let encoded = serde_json::to_vec(&sweep).unwrap();
1344+
assert!(encoded.len() <= MAX_HOSTED_SLACK_CHECKPOINT_BYTES_V1);
1345+
assert!(decode_hosted_slack_poll_checkpoint_v2(&encoded).is_err());
1346+
let output = sweep.completed_output().unwrap();
1347+
assert!(output.operational_status.coverage_complete);
1348+
assert_eq!(
1349+
output.operational_status.freshness_state,
1350+
locality_protocol::ReplicaFreshnessState::Fresh
1351+
);
1352+
assert_eq!(
1353+
output
1354+
.snapshot
1355+
.messages()
1356+
.iter()
1357+
.filter(|message| message.text() == "late near sweep end")
1358+
.count(),
1359+
1
1360+
);
1361+
}
1362+
12501363
#[test]
12511364
fn v1_poll_enums_remain_exhaustive_and_source_compatible() {
12521365
fn kind_name(kind: HostedSlackPollKindV1) -> &'static str {

0 commit comments

Comments
 (0)