Skip to content

[BUG](worker): Revert the async fn window widening from #7621 #7684

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dbeglord merged 1 commit into from
Sep 8, 2026
+18 −150
Merged

[BUG](worker): Revert the async fn window widening from #7621 #7684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 3 additions & 130 deletions rust/worker/src/execution/orchestration/async_function_boundary.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,26 +54,14 @@ pub(crate) fn resolve_boundary_plan_from_version_file(
})
.collect::<Vec<_>>();

// Walk versions newest -> oldest. Boundaries above the completion offset
// are visited furthest-first, so the first one whose window fits
// max_compaction_size is the widest eligible target. Tracking the nearest
// boundary as well preserves the oversized-window error below when no
// boundary fits.
let mut historical_version = None;
let mut next_boundary = None;
let mut furthest_fitting_boundary = None;
for (version, log_position) in version_infos.into_iter().rev() {
if log_position <= completion_offset {
historical_version = Some((version, log_position));
break;
}

if furthest_fitting_boundary.is_none()
&& usize::try_from(log_position - completion_offset)
.is_ok_and(|window| window <= max_compaction_size)
{
furthest_fitting_boundary = Some(log_position);
}
next_boundary = Some(log_position);
}

Expand All @@ -90,13 +78,7 @@ pub(crate) fn resolve_boundary_plan_from_version_file(
None => None,
};

// Prefer the furthest boundary that fits: one run then covers every
// compaction between the completion offset and that boundary, instead of
// draining a backlog one small compaction window at a time. Skipped
// intermediate boundaries are safe — the completion offset advances to a
// real boundary either way, and the work queue retires stale entries by
// offset comparison.
let target_log_position = furthest_fitting_boundary.or(next_boundary).ok_or_else(|| {
let target_log_position = next_boundary.ok_or_else(|| {
format!(
"async fn completion offset {} has no next compaction boundary",
completion_offset
Expand Down Expand Up @@ -211,7 +193,7 @@ mod tests {
}

#[test]
fn completion_offset_zero_uses_empty_state_and_widest_fitting_boundary() {
fn completion_offset_zero_uses_empty_state_and_first_boundary() {
let record_segment = test_record_segment();
let version_file = CollectionVersionFile {
version_history: Some(CollectionVersionHistory {
Expand All @@ -227,122 +209,13 @@ mod tests {
resolve_boundary_plan_from_version_file(Some(&version_file), 0, 1024, &record_segment)
.unwrap();

assert_eq!(plan.target_log_position, 150);
assert_eq!(plan.target_log_position, 100);
assert!(
plan.historical_record_segment.is_none(),
"completion offset zero should use the empty pre-compaction state"
);
}

#[test]
fn picks_furthest_boundary_that_fits_max_compaction_size() {
let record_segment = test_record_segment();
let version_file = CollectionVersionFile {
version_history: Some(CollectionVersionHistory {
versions: vec![
version_info(1, 100, record_segment.id, "record/v100"),
version_info(2, 150, record_segment.id, "record/v150"),
version_info(3, 200, record_segment.id, "record/v200"),
],
}),
..Default::default()
};

let plan = resolve_boundary_plan_from_version_file(
Some(&version_file),
100,
1024,
&record_segment,
)
.unwrap();

assert_eq!(plan.target_log_position, 200);
assert_eq!(
plan.historical_record_segment.unwrap().file_path["offset_id_to_data"],
vec!["record/v100".to_string()]
);
}

#[test]
fn skips_boundaries_wider_than_max_compaction_size() {
let record_segment = test_record_segment();
let version_file = CollectionVersionFile {
version_history: Some(CollectionVersionHistory {
versions: vec![
version_info(1, 100, record_segment.id, "record/v100"),
version_info(2, 150, record_segment.id, "record/v150"),
version_info(3, 200, record_segment.id, "record/v200"),
version_info(4, 5000, record_segment.id, "record/v5000"),
],
}),
..Default::default()
};

let plan = resolve_boundary_plan_from_version_file(
Some(&version_file),
100,
1000,
&record_segment,
)
.unwrap();

assert_eq!(plan.target_log_position, 200);
}

#[test]
fn errors_when_no_boundary_fits_max_compaction_size() {
let record_segment = test_record_segment();
let version_file = CollectionVersionFile {
version_history: Some(CollectionVersionHistory {
versions: vec![
version_info(1, 100, record_segment.id, "record/v100"),
version_info(2, 5000, record_segment.id, "record/v5000"),
],
}),
..Default::default()
};

let err = resolve_boundary_plan_from_version_file(
Some(&version_file),
100,
1000,
&record_segment,
)
.unwrap_err();

assert!(err.contains("exceeds max_compaction_size"));
}

#[test]
fn deleted_versions_are_not_widened_targets() {
let record_segment = test_record_segment();
let mut deleted_version = version_info(2, 150, record_segment.id, "record/v150");
deleted_version.marked_for_deletion = true;

let version_file = CollectionVersionFile {
version_history: Some(CollectionVersionHistory {
versions: vec![
version_info(1, 100, record_segment.id, "record/v100"),
deleted_version,
version_info(3, 5000, record_segment.id, "record/v5000"),
],
}),
..Default::default()
};

// The only live boundary above the offset (5000) does not fit, and the
// deleted 150 boundary must not be picked in its place.
let err = resolve_boundary_plan_from_version_file(
Some(&version_file),
100,
1000,
&record_segment,
)
.unwrap_err();

assert!(err.contains("exceeds max_compaction_size"));
}

#[test]
fn rejects_non_boundary_completion_offsets_after_first_compaction() {
let record_segment = test_record_segment();
Expand Down
35 changes: 15 additions & 20 deletions rust/worker/src/execution/orchestration/compact.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5096,7 +5096,7 @@ mod tests {

// Run three regular compactions to create boundaries at 99, 199, and 299.
// The async fn-consumer will later read from the historical version at 99
// and fetch through the furthest boundary that fits max_compaction_size.
// and fetch only through the next boundary at 199.
for expected_log_position in [99, 199, 299] {
let compact_result = Box::pin(compact(
system.clone(),
Expand Down Expand Up @@ -5138,11 +5138,10 @@ mod tests {
);
}

// Now simulate fn-consumer execution from the first boundary. With a
// max_compaction_size (1000) wider than the whole backlog, this must:
// Now simulate fn-consumer execution from the first boundary. This must:
// 1. read against the historical version compacted through offset 99
// 2. fetch through the furthest committed boundary (299) in one window
// 3. materialize exactly the 200 records in (99, 299]
// 2. stop fetching at the next committed boundary (199)
// 3. materialize exactly the 100 records in (99, 199]
let mut fn_consumer_context = CompactionContext::new_with_log_offset(
None,
50,
Expand Down Expand Up @@ -5178,9 +5177,8 @@ mod tests {
match fetch_response {
LogFetchOrchestratorResponse::Success(success) => {
assert_eq!(
success.collection_info.pulled_log_offset, 299,
"fn-consumer should fetch through the furthest committed boundary \
that fits max_compaction_size"
success.collection_info.pulled_log_offset, 199,
"fn-consumer should only fetch through the next committed boundary"
);

let total_materialized_records: usize = success
Expand All @@ -5189,8 +5187,8 @@ mod tests {
.map(|batch| batch.result.len())
.sum();
assert_eq!(
total_materialized_records, 200,
"fn-consumer should materialize every record in the widened window"
total_materialized_records, 100,
"fn-consumer should materialize exactly the records between boundaries"
);

for batch in &success.materialized {
Expand All @@ -5210,14 +5208,13 @@ mod tests {

Box::pin(fn_consumer_context.cleanup()).await;

// Repeat from the first boundary with a max_compaction_size of exactly
// one window (100) to prove the cap forces the fetch back to the
// nearest committed boundary when the furthest one does not fit.
// Repeat from the second boundary to prove fn-consumer continues to hop
// exactly one committed compaction window at a time.
let mut second_window_context = CompactionContext::new_with_log_offset(
None,
50,
10,
100,
1000,
50,
log.clone(),
sysdb.clone(),
Expand All @@ -5231,7 +5228,7 @@ mod tests {
None,
None,
None,
99,
199,
);

let second_fetch_response = second_window_context
Expand All @@ -5248,9 +5245,8 @@ mod tests {
match second_fetch_response {
LogFetchOrchestratorResponse::Success(success) => {
assert_eq!(
success.collection_info.pulled_log_offset, 199,
"a tight max_compaction_size must stop the fetch at the nearest \
fitting boundary"
success.collection_info.pulled_log_offset, 299,
"fn-consumer should continue to the next committed boundary on later runs"
);

let total_materialized_records: usize = success
Expand All @@ -5260,8 +5256,7 @@ mod tests {
.sum();
assert_eq!(
total_materialized_records, 100,
"a capped fn-consumer run should materialize exactly one \
compaction window"
"each fn-consumer run should materialize exactly one compaction window"
);

for batch in &success.materialized {
Expand Down
Loading