[ENH](worker): Widen async fn windows to the furthest fitting compaction boundary (#7621)

## Why

An async attached function drains its input collection one compaction
window at a time. Each run reads the records between the function's
completion offset and one compaction boundary, sends them to the
external generation service, waits for it to finish, and advances the
offset to that boundary. When a tenant backfills a large history, the
collection accumulates dozens of small compactions — and the function
replays them one by one, paying a full generation round-trip (minutes
each, dominated by remote wall time) for every ~200-record window.

The resolver currently picks the *nearest* boundary above the completion
offset, so the number of round-trips equals the number of compactions,
no matter how small each one is. In production we have observed exactly
this shape: a backfill of tens of thousands of records drains in
hundreds of serial multi-minute round-trips, taking the better part of a
day for a single collection.

## What

The boundary resolver now targets the *furthest* live compaction
boundary whose window still fits within `max_compaction_size`, and falls
back to the nearest boundary (preserving the existing oversized-window
error) when none fits. One run then covers every compaction between the
completion offset and that boundary. With a window cap of 10,000
records, a dense backlog of ~200-record compactions drains in up to ~50×
fewer generation round-trips.

No other component changes:

- The plan already only carries an upper bound for the log read and the
as-of record segment at the completion offset; nothing downstream
requires the target to be the immediately-next version
(`log_fetch_orchestrator` sets
`pulled_log_offset`/`log_upper_bound_offset` from the plan, and the
completion offset advances to exactly that target via
`resolve_pulled_log_offset`).
- Skipped intermediate boundaries retire safely in the work queue:
`CheckInvocationStatus` marks any queued entry done once the function's
completion offset passes it.
- Log retention already holds back garbage collection to the minimum
attached-function completion offset
(`fetch_min_attached_function_completion_offset` in the garbage
collector), and a widened window reads no older logs than today — same
start, further end.
- The HTTP generation executor already splits any window into multiple
requests capped at 16MB / `batch_size` (`batch_requests` in
`http_generate.rs`), so window size and request payload size stay
decoupled.

Changed:
`rust/worker/src/execution/orchestration/async_function_boundary.rs`
(`resolve_boundary_plan_from_version_file`).

## Tests

`cargo test -p worker --lib async_function_boundary` — 10 passed. New
cases: furthest boundary picked when several fit; oversized boundaries
skipped in favor of the widest fitting one; error unchanged when no
boundary fits; deleted versions never become widened targets;
offset-zero backfill now targets the widest fitting boundary.

## Future work: bounded per-function pipelining

Fattened windows fix the round-trips-per-record ratio but keep one
window in flight per function, so remote generation latency still
serializes a single tenant's drain. Pipelining K windows per function is
the next lever (a large backfill's drain time drops roughly by K), but
it is not a local change: concurrent runs resolve from the same
persisted completion offset today (they would process the same window,
not consecutive ones), the offset advance would need low-water-mark
semantics in sysdb to tolerate out-of-order completion, and the
generation service's tolerance for out-of-order windows is a contract
that lives outside this repo. Deferred until those are settled.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: dbeglord

rust/index/src/quantization/single_bit.rs

@@ -218,17 +218,19 @@ impl<T: AsRef<[u8]>> Code<1, T> {
             let p2 = &qq.bit_planes[2 * pb..3 * pb];
             let p3 = &qq.bit_planes[3 * pb..4 * pb];
             let (mut pop0, mut pop1, mut pop2, mut pop3) = (0u32, 0u32, 0u32, 0u32);
-            for (x_chunk, (((q0, q1), q2), q3)) in packed.chunks_exact(8).zip(
-                p0.chunks_exact(8)
-                    .zip(p1.chunks_exact(8))
-                    .zip(p2.chunks_exact(8))
-                    .zip(p3.chunks_exact(8)),
+            for (x_chunk, (((q0, q1), q2), q3)) in packed.as_chunks::<8>().0.iter().zip(
+                p0.as_chunks::<8>()
+                    .0
+                    .iter()
+                    .zip(p1.as_chunks::<8>().0.iter())
+                    .zip(p2.as_chunks::<8>().0.iter())
+                    .zip(p3.as_chunks::<8>().0.iter()),
             ) {
-                let x = u64::from_le_bytes(x_chunk.try_into().unwrap());
-                pop0 += (x & u64::from_le_bytes(q0.try_into().unwrap())).count_ones();
-                pop1 += (x & u64::from_le_bytes(q1.try_into().unwrap())).count_ones();
-                pop2 += (x & u64::from_le_bytes(q2.try_into().unwrap())).count_ones();
-                pop3 += (x & u64::from_le_bytes(q3.try_into().unwrap())).count_ones();
+                let x = u64::from_le_bytes(*x_chunk);
+                pop0 += (x & u64::from_le_bytes(*q0)).count_ones();
+                pop1 += (x & u64::from_le_bytes(*q1)).count_ones();
+                pop2 += (x & u64::from_le_bytes(*q2)).count_ones();
+                pop3 += (x & u64::from_le_bytes(*q3)).count_ones();
             }
             pop0 + (pop1 << 1) + (pop2 << 2) + (pop3 << 3)
         };
@@ -383,10 +385,13 @@ impl Code<1, Vec<u8>> {
         // 16 elements = 64 bytes = one cache line = four NEON / one AVX-512 load.
         // 4 inner loops of 4: alternate between chains (a, b, a, b)
         // so each chain's additions are independent and the OoO core can pipeline them.
-        for (out_pair, (emb_chunk, cen_chunk)) in packed
-            .chunks_exact_mut(2)
-            .zip(embedding.chunks_exact(16).zip(centroid.chunks_exact(16)))
-        {
+        for (out_pair, (emb_chunk, cen_chunk)) in packed.as_chunks_mut::<2>().0.iter_mut().zip(
+            embedding
+                .as_chunks::<16>()
+                .0
+                .iter()
+                .zip(centroid.as_chunks::<16>().0.iter()),
+        ) {
             let mut byte_lo = 0u8;
             let mut byte_hi = 0u8;
 
@@ -495,11 +500,13 @@ fn hamming_distance(a: &[u8], b: &[u8]) -> u32 {
     if let Some(bits) = <u8 as BinarySimilarity>::hamming(a, b) {
         bits as u32
     } else {
-        a.chunks_exact(8)
-            .zip(b.chunks_exact(8))
+        a.as_chunks::<8>()
+            .0
+            .iter()
+            .zip(b.as_chunks::<8>().0.iter())
             .map(|(lhs, rhs)| {
-                let lhs = u64::from_le_bytes(lhs.try_into().unwrap());
-                let rhs = u64::from_le_bytes(rhs.try_into().unwrap());
+                let lhs = u64::from_le_bytes(*lhs);
+                let rhs = u64::from_le_bytes(*rhs);
                 (lhs ^ rhs).count_ones()
             })
             .sum()
@@ -570,15 +577,16 @@ impl QuantizedQuery {
         };
 
         // Single fused pass: quantize each element, accumulate sum, and scatter
-        // bits into a flat bit-plane buffer via chunks_exact(8). The exact-chunk
+        // bits into a flat bit-plane buffer via as_chunks::<8>(). The exact-chunk
         // guarantee lets LLVM eliminate bounds checks and generate tighter code
         // (44% faster than chunks(8) on Apple M-series).
         //
         // Layout: plane j occupies bit_planes[j*padded_bytes .. (j+1)*padded_bytes].
         let inv_delta = 1.0 / delta;
         let mut bit_planes = vec![0u8; B_Q as usize * padded_bytes];
         let mut sum_q_u = 0u32;
-        for (byte_idx, chunk) in r_q.chunks_exact(8).enumerate() {
+        let (r_q_chunks, r_q_rem) = r_q.as_chunks::<8>();
+        for (byte_idx, chunk) in r_q_chunks.iter().enumerate() {
             let (mut b0, mut b1, mut b2, mut b3) = (0u8, 0u8, 0u8, 0u8);
             for (bit, &v) in chunk.iter().enumerate() {
                 let qu = (((v - v_l) * inv_delta).round() as u32).min(max_val);
@@ -594,11 +602,10 @@ impl QuantizedQuery {
             bit_planes[3 * padded_bytes + byte_idx] = b3;
         }
         // Handle remainder for dim not divisible by 8.
-        let rem = r_q.chunks_exact(8).remainder();
-        if !rem.is_empty() {
+        if !r_q_rem.is_empty() {
             let byte_idx = r_q.len() / 8;
             let (mut b0, mut b1, mut b2, mut b3) = (0u8, 0u8, 0u8, 0u8);
-            for (bit, &v) in rem.iter().enumerate() {
+            for (bit, &v) in r_q_rem.iter().enumerate() {
                 let qu = (((v - v_l) * inv_delta).round() as u32).min(max_val);
                 sum_q_u += qu;
                 b0 |= (((qu) & 1) as u8) << bit;

rust/types/src/base64_decode.rs

@@ -103,12 +103,9 @@ pub fn decode_base64_embedding(base64_str: &String) -> Result<Vec<f32>, Base64De
     }
 
     let mut floats = Vec::with_capacity(float_count);
-    for (embedding_index, chunk) in bytes.chunks_exact(4).enumerate() {
-        let float_bytes: [u8; 4] = chunk
-            .try_into()
-            .map_err(|_| Base64DecodeError::EmbeddingConversionFailed { embedding_index })?;
+    for (embedding_index, chunk) in bytes.as_chunks::<4>().0.iter().enumerate() {
         // handles little endian encoding
-        let f = f32::from_le_bytes(float_bytes);
+        let f = f32::from_le_bytes(*chunk);
         if !f.is_finite() {
             return Err(Base64DecodeError::NonFiniteFloatValue {
                 embedding_index,

rust/types/src/sparse_posting_block.rs

@@ -317,8 +317,10 @@ impl SparsePostingBlock {
         let weight_start = Self::body_weight_offset(num_entries, bits_per_delta);
         let weight_bytes = &raw_body[weight_start..weight_start + num_entries * 2];
         let values: Vec<f32> = weight_bytes
-            .chunks_exact(2)
-            .map(|b| f16::from_le_bytes([b[0], b[1]]).to_f32())
+            .as_chunks::<2>()
+            .0
+            .iter()
+            .map(|b| f16::from_le_bytes(*b).to_f32())
             .collect();
 
         Decompressed { offsets, values }
@@ -975,8 +977,8 @@ pub fn convert_f16_to_f32(f16_bytes: &[u8], out: &mut [f32]) {
 
 /// Scalar f16→f32 conversion via the `half` crate.
 pub fn convert_f16_to_f32_scalar(f16_bytes: &[u8], out: &mut [f32]) {
-    for (o, chunk) in out.iter_mut().zip(f16_bytes.chunks_exact(2)) {
-        *o = f16::from_le_bytes([chunk[0], chunk[1]]).to_f32();
+    for (o, chunk) in out.iter_mut().zip(f16_bytes.as_chunks::<2>().0) {
+        *o = f16::from_le_bytes(*chunk).to_f32();
     }
 }
 

rust/types/src/where_parsing.rs

@@ -258,14 +258,13 @@ pub fn parse_where(json_payload: &Value) -> Result<Where, WhereValidationError>
         }
         let (operator, operand) = value_obj.iter().next().unwrap();
         if operand.is_array() {
-            let set_operator;
-            if operator == "$in" {
-                set_operator = crate::SetOperator::In;
+            let set_operator = if operator == "$in" {
+                crate::SetOperator::In
             } else if operator == "$nin" {
-                set_operator = crate::SetOperator::NotIn;
+                crate::SetOperator::NotIn
             } else {
                 return Err(WhereValidationError::WhereClause);
-            }
+            };
             let operand = operand.as_array().unwrap();
             if operand.is_empty() {
                 return Err(WhereValidationError::WhereClause);
@@ -374,14 +373,13 @@ pub fn parse_where(json_payload: &Value) -> Result<Where, WhereValidationError>
                     pattern: operand_str.to_string(),
                 }));
             }
-            let operator_type;
-            if operator == "$eq" {
-                operator_type = PrimitiveOperator::Equal;
+            let operator_type = if operator == "$eq" {
+                PrimitiveOperator::Equal
             } else if operator == "$ne" {
-                operator_type = PrimitiveOperator::NotEqual;
+                PrimitiveOperator::NotEqual
             } else {
                 return Err(WhereValidationError::WhereClause);
-            }
+            };
             return Ok(Where::Metadata(MetadataExpression {
                 key: key.clone(),
                 comparison: crate::MetadataComparison::Primitive(
@@ -405,14 +403,13 @@ pub fn parse_where(json_payload: &Value) -> Result<Where, WhereValidationError>
                     ),
                 }));
             }
-            let operator_type;
-            if operator == "$eq" {
-                operator_type = PrimitiveOperator::Equal;
+            let operator_type = if operator == "$eq" {
+                PrimitiveOperator::Equal
             } else if operator == "$ne" {
-                operator_type = PrimitiveOperator::NotEqual;
+                PrimitiveOperator::NotEqual
             } else {
                 return Err(WhereValidationError::WhereClause);
-            }
+            };
             return Ok(Where::Metadata(MetadataExpression {
                 key: key.clone(),
                 comparison: crate::MetadataComparison::Primitive(
@@ -436,22 +433,21 @@ pub fn parse_where(json_payload: &Value) -> Result<Where, WhereValidationError>
                     ),
                 }));
             }
-            let operator_type;
-            if operator == "$eq" {
-                operator_type = PrimitiveOperator::Equal;
+            let operator_type = if operator == "$eq" {
+                PrimitiveOperator::Equal
             } else if operator == "$ne" {
-                operator_type = PrimitiveOperator::NotEqual;
+                PrimitiveOperator::NotEqual
             } else if operator == "$lt" {
-                operator_type = PrimitiveOperator::LessThan;
+                PrimitiveOperator::LessThan
             } else if operator == "$lte" {
-                operator_type = PrimitiveOperator::LessThanOrEqual;
+                PrimitiveOperator::LessThanOrEqual
             } else if operator == "$gt" {
-                operator_type = PrimitiveOperator::GreaterThan;
+                PrimitiveOperator::GreaterThan
             } else if operator == "$gte" {
-                operator_type = PrimitiveOperator::GreaterThanOrEqual;
+                PrimitiveOperator::GreaterThanOrEqual
             } else {
                 return Err(WhereValidationError::WhereClause);
-            }
+            };
             return Ok(Where::Metadata(MetadataExpression {
                 key: key.clone(),
                 comparison: crate::MetadataComparison::Primitive(
@@ -475,22 +471,21 @@ pub fn parse_where(json_payload: &Value) -> Result<Where, WhereValidationError>
                     ),
                 }));
             }
-            let operator_type;
-            if operator == "$eq" {
-                operator_type = PrimitiveOperator::Equal;
+            let operator_type = if operator == "$eq" {
+                PrimitiveOperator::Equal
             } else if operator == "$ne" {
-                operator_type = PrimitiveOperator::NotEqual;
+                PrimitiveOperator::NotEqual
             } else if operator == "$lt" {
-                operator_type = PrimitiveOperator::LessThan;
+                PrimitiveOperator::LessThan
             } else if operator == "$lte" {
-                operator_type = PrimitiveOperator::LessThanOrEqual;
+                PrimitiveOperator::LessThanOrEqual
             } else if operator == "$gt" {
-                operator_type = PrimitiveOperator::GreaterThan;
+                PrimitiveOperator::GreaterThan
             } else if operator == "$gte" {
-                operator_type = PrimitiveOperator::GreaterThanOrEqual;
+                PrimitiveOperator::GreaterThanOrEqual
             } else {
                 return Err(WhereValidationError::WhereClause);
-            }
+            };
             return Ok(Where::Metadata(MetadataExpression {
                 key: key.clone(),
                 comparison: crate::MetadataComparison::Primitive(