diff --git a/.ci-repro-marker b/.ci-repro-marker
new file mode 100644
index 00000000000..d0ff51d0577
--- /dev/null
+++ b/.ci-repro-marker
@@ -0,0 +1,9 @@
+reproducibility demo marker
+run 2
+run 3
+run 4
+run 5
+repro iteration A
+repro iteration B
+repro iteration C
+repro iteration D
diff --git a/.github/actions/setup-nativelink-cloud/action.yaml b/.github/actions/setup-nativelink-cloud/action.yaml
index 5165e4dd3f1..f259a5d15a0 100644
--- a/.github/actions/setup-nativelink-cloud/action.yaml
+++ b/.github/actions/setup-nativelink-cloud/action.yaml
@@ -77,7 +77,7 @@ inputs:
outputs:
enabled:
- description: "'true' when a user.bazelrc was written, else 'false'."
+ description: "'true' when remote cache or execution is enabled, else 'false'."
value: ${{ steps.configure.outputs.enabled }}
runs:
diff --git a/.github/workflows/native-bazel.yaml b/.github/workflows/native-bazel.yaml
index fd60d6eb788..376fea249fb 100644
--- a/.github/workflows/native-bazel.yaml
+++ b/.github/workflows/native-bazel.yaml
@@ -49,18 +49,16 @@ jobs:
- name: Free disk space
uses: ./.github/actions/free-disk
- - name: Setup Bazel
- uses: >- # v0.19.0
- bazel-contrib/setup-bazel@c5acdfb288317d0b5c0bbd7a396a3dc868bb0f86
- with:
- bazelisk-cache: true
- repository-cache: true
- disk-cache: ${{ github.workflow }}-${{ matrix.os }}-bazel-${{ matrix.bazel_version }}
-
- name: Setup NativeLink Cloud
+ id: setup-nativelink-cloud
uses: ./.github/actions/setup-nativelink-cloud
with:
- ci-mode: ${{ vars.NATIVELINK_CI_MODE }}
+ # Default to 'cache' when the repo variable is unset so forks and
+ # untrusted PRs still get the committed anonymous read-only cache
+ # (cred_kind=public: cache_read only, --remote_upload_local_results
+ # =false, no BES). An explicit 'off' still disables; trusted lanes
+ # with secrets are unaffected.
+ ci-mode: ${{ vars.NATIVELINK_CI_MODE || 'cache' }}
claim-base: ${{ secrets.NATIVELINK_STAGING_CLAIM_BASE }}
bes-results-url: ${{ secrets.NATIVELINK_STAGING_BES_RESULTS_URL }}
api-key: >-
@@ -72,6 +70,16 @@ jobs:
# non-Linux runners, so macOS lanes fall back to cache.
exec: ${{ runner.os == 'Linux' && 'on' || 'off' }}
+ - name: Setup Bazel
+ uses: >- # v0.19.0
+ bazel-contrib/setup-bazel@c5acdfb288317d0b5c0bbd7a396a3dc868bb0f86
+ with:
+ bazelisk-cache: true
+ repository-cache: true
+ # Only enable when we don't have remote cache, otherwise it just gets very slow
+ # and enormous
+ disk-cache: ${{ steps.setup-nativelink-cloud.outputs.enabled == 'false'}}
+
- name: Show Bazel version
run: bazel --version
@@ -119,18 +127,16 @@ jobs:
with:
compose-file: src/bin/docker-compose.store-tester.yaml
- - name: Setup Bazel
- uses: >- # v0.19.0
- bazel-contrib/setup-bazel@c5acdfb288317d0b5c0bbd7a396a3dc868bb0f86
- with:
- bazelisk-cache: true
- repository-cache: true
- disk-cache: ${{ github.workflow }}-ubuntu-24.04
-
- name: Setup NativeLink Cloud
+ id: setup-nativelink-cloud
uses: ./.github/actions/setup-nativelink-cloud
with:
- ci-mode: ${{ vars.NATIVELINK_CI_MODE }}
+ # Default to 'cache' when the repo variable is unset so forks and
+ # untrusted PRs still get the committed anonymous read-only cache
+ # (cred_kind=public: cache_read only, --remote_upload_local_results
+ # =false, no BES). An explicit 'off' still disables; trusted lanes
+ # with secrets are unaffected.
+ ci-mode: ${{ vars.NATIVELINK_CI_MODE || 'cache' }}
claim-base: ${{ secrets.NATIVELINK_STAGING_CLAIM_BASE }}
bes-results-url: ${{ secrets.NATIVELINK_STAGING_BES_RESULTS_URL }}
api-key: >-
@@ -142,6 +148,16 @@ jobs:
# non-Linux runners, so macOS lanes fall back to cache.
exec: ${{ runner.os == 'Linux' && 'on' || 'off' }}
+ - name: Setup Bazel
+ uses: >- # v0.19.0
+ bazel-contrib/setup-bazel@c5acdfb288317d0b5c0bbd7a396a3dc868bb0f86
+ with:
+ bazelisk-cache: true
+ repository-cache: true
+ # Only enable when we don't have remote cache, otherwise it just gets very slow
+ # and enormous
+ disk-cache: ${{ steps.setup-nativelink-cloud.outputs.enabled == 'false'}}
+
- name: Run Store tester with sentinel
run: |
bazel run //:redis_store_tester \
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b8a400a0769..c5853fcf257 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,53 @@
All notable changes to this project will be documented in this file.
+## [1.7.3](https://github.com/TraceMachina/nativelink/compare/v1.7.2..v1.7.3) - 2026-10-01
+
+
+
+### โฐ๏ธ Features
+
+- Support queueing-service through an opt-in gRPC event sink ([#2811](https://github.com/TraceMachina/nativelink/issues/2811)) - ([181a1d3](https://github.com/TraceMachina/nativelink/commit/181a1d34a116aafd93f7b735f224a7e31f314b54))
+- bound the input fetch with max_download_timeout and let a kill reach it ([#2846](https://github.com/TraceMachina/nativelink/issues/2846)) - ([82240c6](https://github.com/TraceMachina/nativelink/commit/82240c640970e8341f6358404456845aad17df47))
+- read capacity and free memory from the worker's own cgroup directory ([#2845](https://github.com/TraceMachina/nativelink/issues/2845)) - ([b4a5468](https://github.com/TraceMachina/nativelink/commit/b4a5468248f4699fcd1ad928e3ea65167e92af86))
+- Historical resource scheduler: hints by digest keys, size classes with a cold-start policy, and where each reservation came from ([#2832](https://github.com/TraceMachina/nativelink/issues/2832)) - ([59cbad2](https://github.com/TraceMachina/nativelink/commit/59cbad26b2a793e06d5e436b3e984f3d1c2df74a))
+- Persistent workers: a configurable pool with an idle sweeper, the action environment, namespaces, measured usage, and a wait at the cap ([#2830](https://github.com/TraceMachina/nativelink/issues/2830)) - ([035dd63](https://github.com/TraceMachina/nativelink/commit/035dd631650a30334db2f5d062dbde8ff3cde301))
+- memory escalation with retry budgets by cause, the whole worker as the last step ([#2829](https://github.com/TraceMachina/nativelink/issues/2829)) - ([3c52ee1](https://github.com/TraceMachina/nativelink/commit/3c52ee1a7026014deef13f92d9cef1e6b474136b))
+- a disk soft limit, and peak_disk_kb from the files an action writes ([#2825](https://github.com/TraceMachina/nativelink/issues/2825)) - ([7bd3c54](https://github.com/TraceMachina/nativelink/commit/7bd3c5424d00bd29f6e3e1516d60105196f439e2))
+- each action gets a private /tmp, its own tmp directory bound over /tmp ([#2824](https://github.com/TraceMachina/nativelink/issues/2824)) - ([c79e3bb](https://github.com/TraceMachina/nativelink/commit/c79e3bbe69e4ecac3e87f9503f47083bffd07fd6))
+- Worker reaps the zombies actions leave behind, and an action's stragglers no longer hold its output open ([#2823](https://github.com/TraceMachina/nativelink/issues/2823)) - ([4fe0a27](https://github.com/TraceMachina/nativelink/commit/4fe0a27ba9e44e02aedf2d197e6f2189c6e543b2))
+- SIGTERM grace before the kill, output kept, a TMPDIR per action, and resource usage with an outcome ([#2822](https://github.com/TraceMachina/nativelink/issues/2822)) - ([9a95e29](https://github.com/TraceMachina/nativelink/commit/9a95e29153f9cbd64208355f211293b97c6d6106))
+- Worker readiness follows scheduler registration ([#2821](https://github.com/TraceMachina/nativelink/issues/2821)) - ([de7a540](https://github.com/TraceMachina/nativelink/commit/de7a540222ca697da3fec83cbb97ba7bcfab039c))
+- Worker advertises CPU and memory from its own cgroup limits ([#2820](https://github.com/TraceMachina/nativelink/issues/2820)) - ([d86ff4f](https://github.com/TraceMachina/nativelink/commit/d86ff4f38ca6e40e09235adfff13cbb036eef9cc))
+- soft memory enforcement against the reservation, and a disk guard before inputs are fetched ([#2812](https://github.com/TraceMachina/nativelink/issues/2812)) - ([fcdccc8](https://github.com/TraceMachina/nativelink/commit/fcdccc81fd4e29e824f8eed733f72988f3b81a18))
+- bound output capture, upload fan-out, precondition scripts and orphaned action directories ([#2810](https://github.com/TraceMachina/nativelink/issues/2810)) - ([f10c7b7](https://github.com/TraceMachina/nativelink/commit/f10c7b70c10a450cefb0062922b33e0840eec8de))
+- kill the whole process group when an action times out or is cancelled ([#2809](https://github.com/TraceMachina/nativelink/issues/2809)) - ([e6aa38a](https://github.com/TraceMachina/nativelink/commit/e6aa38a6b490c281a9df03cf205c951e1a1dbedc))
+- Dispatch is acknowledged or declined by the worker ([#2808](https://github.com/TraceMachina/nativelink/issues/2808)) - ([a21d82c](https://github.com/TraceMachina/nativelink/commit/a21d82c8bdd4225e7f2287ff8018ba243384157b))
+- GoingAway with a drain flag, reconnect backoff with jitter, and a shutdown guard that waits for the drain ([#2807](https://github.com/TraceMachina/nativelink/issues/2807)) - ([2bfa020](https://github.com/TraceMachina/nativelink/commit/2bfa0207f8910d2af4d1b3d2df564a39c1730235))
+- Matching pass: room that opens mid-pass goes to the oldest waiting action, one store read per queued action ([#2806](https://github.com/TraceMachina/nativelink/issues/2806)) - ([2762510](https://github.com/TraceMachina/nativelink/commit/276251061994325162b96a79e12dca3b4dce9e20))
+- Redis scheduler store: RESP3 cursor pages, the retention default, an unreadable row skipped, a wait for the new index ([#2803](https://github.com/TraceMachina/nativelink/issues/2803)) - ([5ef7a83](https://github.com/TraceMachina/nativelink/commit/5ef7a8315d472f92dbb785eb82dfe8bd85968e06))
+- least_loaded and best_fit allocation strategies, typed properties cached per pass ([#2805](https://github.com/TraceMachina/nativelink/issues/2805)) - ([a7cc44d](https://github.com/TraceMachina/nativelink/commit/a7cc44da142e0a8a136176213fa48f2e71464145))
+- Admin API: list connected workers and queued demand ([#2804](https://github.com/TraceMachina/nativelink/issues/2804)) - ([02a64f2](https://github.com/TraceMachina/nativelink/commit/02a64f2b4c69ba5ae157c8c88f7bf6aad8aaa730))
+- Serve the scheduler queue in order, and let workers report their load on the keepalive ([#2796](https://github.com/TraceMachina/nativelink/issues/2796)) - ([4c06bb8](https://github.com/TraceMachina/nativelink/commit/4c06bb81ec4702d830d8aba7b6f39f225e016aa6))
+- Unsatisfiable actions: the clock follows the fleet's shapes, and a timeout for a fleet with no worker at all ([#2799](https://github.com/TraceMachina/nativelink/issues/2799)) - ([31be880](https://github.com/TraceMachina/nativelink/commit/31be88082f7c376d8b8648b1017d10975215c66a))
+- Scheduler worker state: budget released at completion, pause until the next keepalive, retry cap fails with FailedPrecondition ([#2798](https://github.com/TraceMachina/nativelink/issues/2798)) - ([71bb05a](https://github.com/TraceMachina/nativelink/commit/71bb05a0eedb89b69a4d8b5b5b562bea261644cc))
+- Worker liveness: any message counts, a 10s default, a dead keepalive reconnects, SIGINT drains ([#2797](https://github.com/TraceMachina/nativelink/issues/2797)) - ([2f54cc3](https://github.com/TraceMachina/nativelink/commit/2f54cc3bcb9548b83eedc447bfa02faca8120374))
+
+### ๐ Bug Fixes
+
+- a stalled scheduler connection reconnects instead of leaving the worker evicted and idle ([#2856](https://github.com/TraceMachina/nativelink/issues/2856)) - ([389cf73](https://github.com/TraceMachina/nativelink/commit/389cf73574e82ebee101beacea4453424908b06d))
+- a fast tier that ends short on a digest resumes from the slow store, and a missing input names its digest so the client re-uploads it ([#2855](https://github.com/TraceMachina/nativelink/issues/2855)) - ([bd1bb34](https://github.com/TraceMachina/nativelink/commit/bd1bb3402e03fd6b377ec7166dd3ba6f0c1b4416))
+- Give a live worker a grace period to report its own timeout ([#2826](https://github.com/TraceMachina/nativelink/issues/2826)) - ([5b4e555](https://github.com/TraceMachina/nativelink/commit/5b4e555070fe5e1b3e8709d11d03c1fc6812023b))
+- a read that stops short of the digest's size is an error, not EOF ([#2844](https://github.com/TraceMachina/nativelink/issues/2844)) - ([fd6530e](https://github.com/TraceMachina/nativelink/commit/fd6530ea797355506dc4f2ca6e730c398f8900cc))
+- build the cleanup mark's guard after the lock is released ([#2843](https://github.com/TraceMachina/nativelink/issues/2843)) - ([9a03932](https://github.com/TraceMachina/nativelink/commit/9a03932c0fd5b38150a865770c3222737b306836))
+- Scheduler skips and counts a queue entry it cannot read ([#2795](https://github.com/TraceMachina/nativelink/issues/2795)) - ([4001f52](https://github.com/TraceMachina/nativelink/commit/4001f5256eccfbfa1f09b849a7f03e4092c4f9bc))
+- Sorted index reads return the whole set, not ten rows ([#2794](https://github.com/TraceMachina/nativelink/issues/2794)) - ([7b2c02d](https://github.com/TraceMachina/nativelink/commit/7b2c02db4993bffb6073d408d7955d34e7882e1f))
+
+### ๐งช Testing & CI
+
+- Give the oversized-upload store tests the 30 s deadline the third already has ([#2818](https://github.com/TraceMachina/nativelink/issues/2818)) - ([3ff63f1](https://github.com/TraceMachina/nativelink/commit/3ff63f17f52372c7ee7459d4f5f7690713fa6e5c))
+- Give trivy three vulnerability database mirrors to fall through ([#2801](https://github.com/TraceMachina/nativelink/issues/2801)) - ([a9c6801](https://github.com/TraceMachina/nativelink/commit/a9c6801f4e374016734811eb35231e14303778c8))
+
## [1.7.2](https://github.com/TraceMachina/nativelink/compare/v1.7.1..v1.7.2) - 2026-09-25
### Major changes
diff --git a/Cargo.lock b/Cargo.lock
index ec5790ff881..89618fc6192 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2926,7 +2926,7 @@ checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084"
[[package]]
name = "nativelink"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"async-lock",
"axum",
@@ -2957,7 +2957,7 @@ dependencies = [
[[package]]
name = "nativelink-config"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"byte-unit",
"humantime",
@@ -2976,7 +2976,7 @@ dependencies = [
[[package]]
name = "nativelink-error"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"base64 0.22.1",
"mongodb",
@@ -3000,7 +3000,7 @@ dependencies = [
[[package]]
name = "nativelink-macro"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"proc-macro2",
"quote",
@@ -3009,7 +3009,7 @@ dependencies = [
[[package]]
name = "nativelink-metric"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"async-lock",
"nativelink-metric-macro-derive",
@@ -3020,7 +3020,7 @@ dependencies = [
[[package]]
name = "nativelink-metric-macro-derive"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"proc-macro2",
"quote",
@@ -3029,7 +3029,7 @@ dependencies = [
[[package]]
name = "nativelink-proto"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"derive_more 2.1.0",
"prost",
@@ -3041,7 +3041,7 @@ dependencies = [
[[package]]
name = "nativelink-redis-tester"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"either",
"nativelink-util",
@@ -3054,7 +3054,7 @@ dependencies = [
[[package]]
name = "nativelink-scheduler"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"async-lock",
"async-trait",
@@ -3091,7 +3091,7 @@ dependencies = [
[[package]]
name = "nativelink-service"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"async-lock",
"async-trait",
@@ -3135,7 +3135,7 @@ dependencies = [
[[package]]
name = "nativelink-store"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"async-lock",
"async-trait",
@@ -3214,7 +3214,7 @@ dependencies = [
[[package]]
name = "nativelink-util"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"anyhow",
"async-trait",
@@ -3277,7 +3277,7 @@ dependencies = [
[[package]]
name = "nativelink-worker"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"async-lock",
"async-trait",
@@ -4525,9 +4525,9 @@ dependencies = [
[[package]]
name = "serial_test"
-version = "3.5.0"
+version = "4.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "699f4197115b8a7e7ff19c9a315a4bd6fffec26cc4626ef45ecaea389e081c6d"
+checksum = "a6df5ed973ad8d834e09f824f9e9f449af6b9a3745f78dec7cc752770bd3bf11"
dependencies = [
"futures-executor",
"futures-util",
@@ -4538,13 +4538,13 @@ dependencies = [
[[package]]
name = "serial_test_derive"
-version = "3.5.0"
+version = "4.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94e153fc76e1c6a068703d6d29c508a0b15c061c4b7e43da59cc097bc342673c"
+checksum = "a22144e767da4ddd8416dbf383700542ffd8a5dc493dfecedfe1fe3ad03c98ae"
dependencies = [
"proc-macro2",
"quote",
- "syn 2.0.117",
+ "syn 3.0.1",
]
[[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 1643d711a50..b3c9f51bfb4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -11,7 +11,7 @@ resolver = "2"
edition = "2024"
name = "nativelink"
rust-version = "1.97.1"
-version = "1.7.2"
+version = "1.7.3"
[profile.release]
lto = true
diff --git a/MODULE.bazel b/MODULE.bazel
index d05045b0bf7..04f49a90337 100644
--- a/MODULE.bazel
+++ b/MODULE.bazel
@@ -1,6 +1,6 @@
module(
name = "nativelink",
- version = "1.7.2",
+ version = "1.7.3",
compatibility_level = 0,
)
diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock
index 257355b8d7b..9c74a820ace 100644
--- a/MODULE.bazel.lock
+++ b/MODULE.bazel.lock
@@ -1145,8 +1145,8 @@
"serde_urlencoded_0.7.1": "{\"dependencies\":[{\"name\":\"form_urlencoded\",\"req\":\"^1\"},{\"name\":\"itoa\",\"req\":\"^1\"},{\"name\":\"ryu\",\"req\":\"^1\"},{\"name\":\"serde\",\"req\":\"^1.0.69\"},{\"kind\":\"dev\",\"name\":\"serde_derive\",\"req\":\"^1\"}],\"features\":{}}",
"serde_with_3.21.0": "{\"dependencies\":[{\"default_features\":false,\"name\":\"base64\",\"optional\":true,\"req\":\"^0.22.1\"},{\"default_features\":false,\"name\":\"bs58\",\"optional\":true,\"req\":\"^0.5.1\"},{\"default_features\":false,\"features\":[\"serde\"],\"name\":\"chrono_0_4\",\"optional\":true,\"package\":\"chrono\",\"req\":\"^0.4.20\"},{\"name\":\"document-features\",\"optional\":true,\"req\":\"^0.2.7\"},{\"kind\":\"dev\",\"name\":\"expect-test\",\"req\":\"^1.5.1\"},{\"kind\":\"dev\",\"name\":\"fnv\",\"req\":\"^1.0.6\"},{\"kind\":\"dev\",\"name\":\"glob\",\"req\":\"^0.3.3\"},{\"default_features\":false,\"features\":[\"serde\"],\"name\":\"hashbrown_0_14\",\"optional\":true,\"package\":\"hashbrown\",\"req\":\"^0.14.0\"},{\"default_features\":false,\"features\":[\"serde\"],\"name\":\"hashbrown_0_15\",\"optional\":true,\"package\":\"hashbrown\",\"req\":\"^0.15.0\"},{\"default_features\":false,\"features\":[\"serde\"],\"name\":\"hashbrown_0_16\",\"optional\":true,\"package\":\"hashbrown\",\"req\":\"^0.16.0\"},{\"default_features\":false,\"features\":[\"serde\"],\"name\":\"hashbrown_0_17\",\"optional\":true,\"package\":\"hashbrown\",\"req\":\"^0.17.0\"},{\"default_features\":false,\"name\":\"hex\",\"optional\":true,\"req\":\"^0.4.3\"},{\"default_features\":false,\"features\":[\"serde-1\"],\"name\":\"indexmap_1\",\"optional\":true,\"package\":\"indexmap\",\"req\":\"^1.8\"},{\"default_features\":false,\"features\":[\"serde\"],\"name\":\"indexmap_2\",\"optional\":true,\"package\":\"indexmap\",\"req\":\"^2.0\"},{\"default_features\":false,\"features\":[\"resolve-file\"],\"kind\":\"dev\",\"name\":\"jsonschema\",\"req\":\"^0.33.0\"},{\"kind\":\"dev\",\"name\":\"mime\",\"req\":\"^0.3.16\"},{\"kind\":\"dev\",\"name\":\"pretty_assertions\",\"req\":\"^1.4.0\"},{\"default_features\":false,\"features\":[\"std\"],\"kind\":\"dev\",\"name\":\"regex\",\"req\":\"^1.12.1\"},{\"kind\":\"dev\",\"name\":\"rmp-serde\",\"req\":\"^1.3.0\"},{\"kind\":\"dev\",\"name\":\"ron\",\"req\":\"^0.12\"},{\"kind\":\"dev\",\"name\":\"rustversion\",\"req\":\"^1.0.22\"},{\"default_features\":false,\"name\":\"schemars_0_8\",\"optional\":true,\"package\":\"schemars\",\"req\":\"^0.8.16\"},{\"kind\":\"dev\",\"name\":\"schemars_0_8\",\"package\":\"schemars\",\"req\":\"^0.8.16\"},{\"default_features\":false,\"name\":\"schemars_0_9\",\"optional\":true,\"package\":\"schemars\",\"req\":\"^0.9.0\"},{\"kind\":\"dev\",\"name\":\"schemars_0_9\",\"package\":\"schemars\",\"req\":\"^0.9.0\"},{\"default_features\":false,\"name\":\"schemars_1\",\"optional\":true,\"package\":\"schemars\",\"req\":\"^1.0.2\"},{\"kind\":\"dev\",\"name\":\"schemars_1\",\"package\":\"schemars\",\"req\":\"^1.0.2\"},{\"default_features\":false,\"features\":[\"derive\"],\"kind\":\"dev\",\"name\":\"serde\",\"req\":\"^1.0.152\"},{\"kind\":\"dev\",\"name\":\"serde-xml-rs\",\"req\":\"^0.8.1\"},{\"default_features\":false,\"features\":[\"result\"],\"name\":\"serde_core\",\"req\":\"^1.0.225\"},{\"default_features\":false,\"name\":\"serde_json\",\"optional\":true,\"req\":\"^1.0.145\"},{\"features\":[\"preserve_order\"],\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.25\"},{\"kind\":\"dev\",\"name\":\"serde_test\",\"req\":\"^1.0.124\"},{\"name\":\"serde_with_macros\",\"optional\":true,\"req\":\"=3.21.0\"},{\"default_features\":false,\"name\":\"smallvec_1\",\"optional\":true,\"package\":\"smallvec\",\"req\":\"^1\"},{\"default_features\":false,\"name\":\"time_0_3\",\"optional\":true,\"package\":\"time\",\"req\":\"~0.3.47\"},{\"kind\":\"dev\",\"name\":\"yaml_serde\",\"req\":\"^0.10.3\"}],\"features\":{\"alloc\":[\"serde_core/alloc\",\"base64?/alloc\",\"bs58?/alloc\",\"chrono_0_4?/alloc\",\"hex?/alloc\",\"serde_json?/alloc\",\"time_0_3?/alloc\"],\"base58\":[\"dep:bs58\",\"alloc\"],\"base64\":[\"dep:base64\",\"alloc\"],\"chrono\":[\"chrono_0_4\"],\"chrono_0_4\":[\"dep:chrono_0_4\"],\"default\":[\"std\",\"macros\"],\"guide\":[\"dep:document-features\",\"macros\",\"std\"],\"hashbrown_0_14\":[\"dep:hashbrown_0_14\",\"alloc\"],\"hashbrown_0_15\":[\"dep:hashbrown_0_15\",\"alloc\"],\"hashbrown_0_16\":[\"dep:hashbrown_0_16\",\"alloc\"],\"hashbrown_0_17\":[\"dep:hashbrown_0_17\",\"alloc\"],\"hex\":[\"dep:hex\",\"alloc\"],\"indexmap\":[\"indexmap_1\"],\"indexmap_1\":[\"dep:indexmap_1\",\"alloc\"],\"indexmap_2\":[\"dep:indexmap_2\",\"alloc\"],\"json\":[\"dep:serde_json\",\"alloc\"],\"macros\":[\"dep:serde_with_macros\"],\"schemars_0_8\":[\"dep:schemars_0_8\",\"std\",\"serde_with_macros?/schemars_0_8\",\"dep:serde_json\"],\"schemars_0_9\":[\"dep:schemars_0_9\",\"alloc\",\"serde_with_macros?/schemars_0_9\",\"dep:serde_json\"],\"schemars_1\":[\"dep:schemars_1\",\"alloc\",\"serde_with_macros?/schemars_1\",\"dep:serde_json\"],\"smallvec_1\":[\"dep:smallvec_1\"],\"std\":[\"alloc\",\"bs58?/std\",\"serde_core/std\",\"chrono_0_4?/clock\",\"chrono_0_4?/std\",\"indexmap_1?/std\",\"indexmap_2?/std\",\"time_0_3?/serde-well-known\",\"time_0_3?/std\",\"schemars_0_9?/std\",\"schemars_1?/std\"],\"time_0_3\":[\"dep:time_0_3\"]}}",
"serde_with_macros_3.21.0": "{\"dependencies\":[{\"name\":\"darling\",\"req\":\"^0.23.0\"},{\"kind\":\"dev\",\"name\":\"expect-test\",\"req\":\"^1.5.1\"},{\"kind\":\"dev\",\"name\":\"glob\",\"req\":\"^0.3.3\"},{\"kind\":\"dev\",\"name\":\"pretty_assertions\",\"req\":\"^1.4.0\"},{\"name\":\"proc-macro2\",\"req\":\"^1.0.1\"},{\"name\":\"quote\",\"req\":\"^1.0.0\"},{\"default_features\":false,\"features\":[\"std\"],\"kind\":\"dev\",\"name\":\"regex\",\"req\":\"^1.12.1\"},{\"kind\":\"dev\",\"name\":\"rustversion\",\"req\":\"^1.0.22\"},{\"features\":[\"derive\"],\"kind\":\"dev\",\"name\":\"serde\",\"req\":\"^1.0.152\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.25\"},{\"features\":[\"extra-traits\",\"full\",\"parsing\"],\"name\":\"syn\",\"req\":\"^2.0.0\"},{\"kind\":\"dev\",\"name\":\"trybuild\",\"req\":\"^1.0.111\"}],\"features\":{\"schemars_0_8\":[],\"schemars_0_9\":[],\"schemars_1\":[]}}",
- "serial_test_3.5.0": "{\"dependencies\":[{\"name\":\"document-features\",\"optional\":true,\"req\":\"^0.2\"},{\"default_features\":false,\"name\":\"env_logger\",\"optional\":true,\"req\":\">=0.6.1\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"fslock\",\"optional\":true,\"req\":\"^0.2\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"futures-executor\",\"optional\":true,\"req\":\"^0.3\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"futures-util\",\"optional\":true,\"req\":\"^0.3\"},{\"default_features\":false,\"features\":[\"use_std\"],\"kind\":\"dev\",\"name\":\"itertools\",\"req\":\">=0.4\"},{\"name\":\"log\",\"optional\":true,\"req\":\">=0.4.4\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"once_cell\",\"req\":\"^1.19\"},{\"default_features\":false,\"name\":\"parking_lot\",\"req\":\"^0.12\"},{\"name\":\"serial_test_derive\",\"req\":\"~3.5.0\"}],\"features\":{\"async\":[\"dep:futures-executor\",\"dep:futures-util\",\"serial_test_derive/async\"],\"default\":[\"logging\",\"async\"],\"docsrs\":[\"dep:document-features\"],\"file_locks\":[\"dep:fslock\"],\"logging\":[\"dep:log\"],\"test_logging\":[\"logging\",\"dep:env_logger\",\"serial_test_derive/test_logging\"]}}",
- "serial_test_derive_3.5.0": "{\"dependencies\":[{\"default_features\":false,\"kind\":\"dev\",\"name\":\"env_logger\",\"req\":\">=0.6.1\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"prettyplease\",\"req\":\"^0.2\"},{\"default_features\":false,\"features\":[\"proc-macro\"],\"name\":\"proc-macro2\",\"req\":\"^1.0.60\"},{\"default_features\":false,\"name\":\"quote\",\"req\":\"^1\"},{\"default_features\":false,\"features\":[\"full\",\"printing\",\"parsing\",\"clone-impls\"],\"name\":\"syn\",\"req\":\"^2\"}],\"features\":{\"async\":[],\"default\":[],\"file_locks\":[],\"test_logging\":[]}}",
+ "serial_test_4.0.1": "{\"dependencies\":[{\"name\":\"document-features\",\"optional\":true,\"req\":\"^0.2\"},{\"default_features\":false,\"name\":\"env_logger\",\"optional\":true,\"req\":\">=0.6.1\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"fslock\",\"optional\":true,\"req\":\"^0.2\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"futures-executor\",\"optional\":true,\"req\":\"^0.3\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"futures-util\",\"optional\":true,\"req\":\"^0.3\"},{\"default_features\":false,\"features\":[\"use_std\"],\"kind\":\"dev\",\"name\":\"itertools\",\"req\":\">=0.4\"},{\"name\":\"log\",\"optional\":true,\"req\":\">=0.4.4\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"once_cell\",\"req\":\"^1.19\"},{\"default_features\":false,\"name\":\"parking_lot\",\"req\":\"^0.12\"},{\"name\":\"serial_test_derive\",\"req\":\"~4.0.1\"}],\"features\":{\"async\":[\"dep:futures-executor\",\"dep:futures-util\",\"serial_test_derive/async\"],\"default\":[\"logging\",\"async\"],\"docsrs\":[\"dep:document-features\"],\"file_locks\":[\"dep:fslock\"],\"logging\":[\"dep:log\"],\"test_logging\":[\"logging\",\"dep:env_logger\",\"serial_test_derive/test_logging\"]}}",
+ "serial_test_derive_4.0.1": "{\"dependencies\":[{\"default_features\":false,\"kind\":\"dev\",\"name\":\"env_logger\",\"req\":\">=0.6.1\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"prettyplease\",\"req\":\"^0.3\"},{\"default_features\":false,\"features\":[\"proc-macro\"],\"name\":\"proc-macro2\",\"req\":\"^1.0.60\"},{\"default_features\":false,\"name\":\"quote\",\"req\":\"^1\"},{\"default_features\":false,\"features\":[\"full\",\"printing\",\"parsing\",\"clone-impls\"],\"name\":\"syn\",\"req\":\"^3\"}],\"features\":{\"async\":[],\"default\":[],\"file_locks\":[],\"test_logging\":[]}}",
"sha1_0.10.6": "{\"dependencies\":[{\"name\":\"cfg-if\",\"req\":\"^1.0\"},{\"name\":\"cpufeatures\",\"req\":\"^0.2\",\"target\":\"cfg(any(target_arch = \\\"aarch64\\\", target_arch = \\\"x86\\\", target_arch = \\\"x86_64\\\"))\"},{\"name\":\"digest\",\"req\":\"^0.10.7\"},{\"features\":[\"dev\"],\"kind\":\"dev\",\"name\":\"digest\",\"req\":\"^0.10.7\"},{\"kind\":\"dev\",\"name\":\"hex-literal\",\"req\":\"^0.2.2\"},{\"name\":\"sha1-asm\",\"optional\":true,\"req\":\"^0.5\",\"target\":\"cfg(any(target_arch = \\\"aarch64\\\", target_arch = \\\"x86\\\", target_arch = \\\"x86_64\\\"))\"}],\"features\":{\"asm\":[\"sha1-asm\"],\"compress\":[],\"default\":[\"std\"],\"force-soft\":[],\"loongarch64_asm\":[],\"oid\":[\"digest/oid\"],\"std\":[\"digest/std\"]}}",
"sha1_0.11.0": "{\"dependencies\":[{\"name\":\"cfg-if\",\"req\":\"^1.0\"},{\"name\":\"cpufeatures\",\"req\":\"^0.3\",\"target\":\"cfg(any(target_arch = \\\"aarch64\\\", target_arch = \\\"x86\\\", target_arch = \\\"x86_64\\\"))\"},{\"name\":\"digest\",\"req\":\"^0.11\"},{\"features\":[\"dev\"],\"kind\":\"dev\",\"name\":\"digest\",\"req\":\"^0.11\"},{\"kind\":\"dev\",\"name\":\"hex-literal\",\"req\":\"^1\"}],\"features\":{\"alloc\":[\"digest/alloc\"],\"default\":[\"alloc\",\"oid\"],\"oid\":[\"digest/oid\"],\"zeroize\":[\"digest/zeroize\"]}}",
"sha1_smol_1.0.1": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"openssl\",\"req\":\"^0.10\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.4\"},{\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0\"}],\"features\":{\"alloc\":[],\"std\":[\"alloc\"]}}",
diff --git a/cliff.toml b/cliff.toml
index 716467aaf87..65d460d3982 100644
--- a/cliff.toml
+++ b/cliff.toml
@@ -72,6 +72,40 @@ commit_parsers = [
# Custom rules for specific commits.
+ { message = "Dispatch is acknowledged or declined by the worker", group = "โฐ๏ธ Features" },
+ { message = "Give a live worker a grace period to report its own timeout", group = "๐ Bug Fixes" },
+ { message = "Give the oversized-upload store tests the 30 s deadline", group = "๐งช Testing & CI" },
+ { message = "Give trivy three vulnerability database mirrors", group = "๐งช Testing & CI" },
+ { message = "GoingAway with a drain flag, reconnect backoff with jitter", group = "โฐ๏ธ Features" },
+ { message = "RESP3 cursor pages, the retention default", group = "โฐ๏ธ Features" },
+ { message = "SIGTERM grace before the kill, output kept", group = "โฐ๏ธ Features" },
+ { message = "Scheduler skips and counts a queue entry it cannot read", group = "๐ Bug Fixes" },
+ { message = "Serve the scheduler queue in order", group = "โฐ๏ธ Features" },
+ { message = "Sorted index reads return the whole set", group = "๐ Bug Fixes" },
+ { message = "Support queueing-service through an opt-in gRPC event sink", group = "โฐ๏ธ Features" },
+ { message = "Worker advertises CPU and memory from its own cgroup limits", group = "โฐ๏ธ Features" },
+ { message = "Worker readiness follows scheduler registration", group = "โฐ๏ธ Features" },
+ { message = "a configurable pool with an idle sweeper", group = "โฐ๏ธ Features" },
+ { message = "a disk soft limit, and peak_disk_kb", group = "โฐ๏ธ Features" },
+ { message = "a fast tier that ends short on a digest resumes from the slow store", group = "๐ Bug Fixes" },
+ { message = "a read that stops short of the digest's size is an error", group = "๐ Bug Fixes" },
+ { message = "a stalled scheduler connection reconnects instead of leaving the worker evicted", group = "๐ Bug Fixes" },
+ { message = "any message counts, a 10s default", group = "โฐ๏ธ Features" },
+ { message = "bound output capture, upload fan-out, precondition scripts", group = "โฐ๏ธ Features" },
+ { message = "bound the input fetch with max_download_timeout", group = "โฐ๏ธ Features" },
+ { message = "budget released at completion, pause until the next keepalive", group = "โฐ๏ธ Features" },
+ { message = "build the cleanup mark's guard after the lock is released", group = "๐ Bug Fixes" },
+ { message = "each action gets a private /tmp", group = "โฐ๏ธ Features" },
+ { message = "hints by digest keys, size classes with a cold-start policy", group = "โฐ๏ธ Features" },
+ { message = "kill the whole process group when an action times out", group = "โฐ๏ธ Features" },
+ { message = "list connected workers and queued demand", group = "โฐ๏ธ Features" },
+ { message = "memory escalation with retry budgets by cause", group = "โฐ๏ธ Features" },
+ { message = "read capacity and free memory from the worker's own cgroup directory", group = "โฐ๏ธ Features" },
+ { message = "reaps the zombies actions leave behind", group = "โฐ๏ธ Features" },
+ { message = "room that opens mid-pass goes to the oldest waiting action", group = "โฐ๏ธ Features" },
+ { message = "soft memory enforcement against the reservation", group = "โฐ๏ธ Features" },
+ { message = "the clock follows the fleet's shapes", group = "โฐ๏ธ Features" },
+
{ message = "Account for block size in filesystem store for eviction purposes", group = "๐ Bug Fixes" },
{ message = "Add ONTAP S3 Store with existence cache", group = "โฐ๏ธ Features" },
{ message = "Add OSFamily property in basic_cas.json", group = "๐ Documentation" },
diff --git a/llms-full.txt b/llms-full.txt
index d2abea0b8e7..3c49ff85bca 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -11416,6 +11416,23 @@ restart.
forward selected headers from the incoming request. Use it to front an existing
cache.
+Development builds also support `grpc.store_type: "event_sink"` for forwarding
+BEP and origin events to a dedicated receiver. This mode accepts string keys
+with exact payload sizes and sends uncompressed `ByteStream.Write` requests.
+The resource hash identifies the event key, while the size describes the event
+bytes. Configure it only as an event publisher's store: reads, existence checks,
+and digest-key writes are rejected. A normal CAS endpoint cannot receive these
+writes. The receiver must acknowledge the full payload size only after its
+durable append completes; NativeLink rejects a different committed size, and
+its BEP response waits for that acknowledgement. A lost acknowledgement can
+cause a retry after the receiver has already persisted an event. The receiver
+must use the stable event key to detect and ignore duplicate events, including
+retries that start a new upload; the upload UUID alone is not an event identity.
+Origin publishing retains its existing bounded retry policy. This transport
+provides neither exactly-once processing nor a durable publisher recovery ledger.
+
+
+
`noop` takes nothing at all and discards everything written to it, reporting
every read as a miss. It is how you deliberately turn a half of something off.
diff --git a/nativelink-config/Cargo.toml b/nativelink-config/Cargo.toml
index d3d52c8e2a3..ff5a5765ae9 100644
--- a/nativelink-config/Cargo.toml
+++ b/nativelink-config/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-config"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-error = { path = "../nativelink-error" }
diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs
index 3e52a8d1741..6fddb93bf70 100644
--- a/nativelink-config/src/cas_server.rs
+++ b/nativelink-config/src/cas_server.rs
@@ -877,10 +877,35 @@ pub struct EndpointConfig {
#[serde(deserialize_with = "convert_string_with_shellexpand")]
pub uri: String,
- /// Timeout in seconds that a request should take.
+ /// Timeout in seconds for connecting to the endpoint and for a request
+ /// to be answered. The worker's messages to the scheduler (keepalives,
+ /// acknowledgements, results) ride on the one `ConnectWorker` stream
+ /// opened at registration, which this does not bound; the `GoingAway`
+ /// sent at shutdown is bounded by it, so a stalled connection cannot
+ /// hold a pod past its grace period. A keepalive goes out every half of
+ /// it.
/// Default: 5 seconds
pub timeout: Option,
+ /// TCP keepalive interval (seconds), probing the socket under the
+ /// scheduler connection so a dead peer is noticed at the OS level.
+ /// If not set or 0, defaults to 30 seconds.
+ #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
+ pub tcp_keepalive_s: u64,
+
+ /// HTTP/2 keepalive interval (seconds): PING frames on the scheduler
+ /// connection, so one that has gone dead underneath fails instead of
+ /// hanging every message on it.
+ /// If not set or 0, defaults to 30 seconds.
+ #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
+ pub http2_keepalive_interval_s: u64,
+
+ /// HTTP/2 keepalive timeout (seconds): a PING unanswered for this long
+ /// ends the connection, and the worker reconnects.
+ /// If not set or 0, defaults to 20 seconds.
+ #[serde(default, deserialize_with = "convert_duration_with_shellexpand")]
+ pub http2_keepalive_timeout_s: u64,
+
/// The TLS configuration to use to connect to the endpoint.
pub tls_config: Option,
}
diff --git a/nativelink-error/Cargo.toml b/nativelink-error/Cargo.toml
index 0a97dfedfb3..fca255e5870 100644
--- a/nativelink-error/Cargo.toml
+++ b/nativelink-error/Cargo.toml
@@ -8,7 +8,7 @@ autoexamples = false
autotests = true
edition = "2024"
name = "nativelink-error"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-metric = { path = "../nativelink-metric" }
diff --git a/nativelink-macro/Cargo.toml b/nativelink-macro/Cargo.toml
index 8954214929e..7684456403f 100644
--- a/nativelink-macro/Cargo.toml
+++ b/nativelink-macro/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-macro"
-version = "1.7.2"
+version = "1.7.3"
[lib]
proc-macro = true
diff --git a/nativelink-metric/Cargo.toml b/nativelink-metric/Cargo.toml
index e14e11f0151..1542fe48dd3 100644
--- a/nativelink-metric/Cargo.toml
+++ b/nativelink-metric/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-metric"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-metric-macro-derive = { path = "nativelink-metric-macro-derive" }
diff --git a/nativelink-metric/nativelink-metric-macro-derive/Cargo.toml b/nativelink-metric/nativelink-metric-macro-derive/Cargo.toml
index 9cf8b22a8ac..a5ef6436211 100644
--- a/nativelink-metric/nativelink-metric-macro-derive/Cargo.toml
+++ b/nativelink-metric/nativelink-metric-macro-derive/Cargo.toml
@@ -1,7 +1,7 @@
[package]
edition = "2024"
name = "nativelink-metric-macro-derive"
-version = "1.7.2"
+version = "1.7.3"
[lib]
proc-macro = true
diff --git a/nativelink-proto/Cargo.toml b/nativelink-proto/Cargo.toml
index e1b95545f10..7f421f771fd 100644
--- a/nativelink-proto/Cargo.toml
+++ b/nativelink-proto/Cargo.toml
@@ -2,7 +2,7 @@
[package]
edition = "2024"
name = "nativelink-proto"
-version = "1.7.2"
+version = "1.7.3"
[lib]
doctest = false # because some of the generated protos have things that look like doctests but break
diff --git a/nativelink-redis-tester/Cargo.toml b/nativelink-redis-tester/Cargo.toml
index 79d0ab2401e..7969bca12a9 100644
--- a/nativelink-redis-tester/Cargo.toml
+++ b/nativelink-redis-tester/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-redis-tester"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-util = { path = "../nativelink-util" }
diff --git a/nativelink-scheduler/Cargo.toml b/nativelink-scheduler/Cargo.toml
index b6747197275..ff752b77941 100644
--- a/nativelink-scheduler/Cargo.toml
+++ b/nativelink-scheduler/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-scheduler"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-config = { path = "../nativelink-config" }
diff --git a/nativelink-scheduler/src/simple_scheduler_state_manager.rs b/nativelink-scheduler/src/simple_scheduler_state_manager.rs
index 664aa1eba4c..79489efc9fe 100644
--- a/nativelink-scheduler/src/simple_scheduler_state_manager.rs
+++ b/nativelink-scheduler/src/simple_scheduler_state_manager.rs
@@ -46,6 +46,34 @@ use super::awaited_action_db::{
};
use crate::worker_registry::{ORPHANED_ACTION_TIMEOUT, SharedWorkerRegistry, WorkerLiveness};
+/// Why the scheduler times out an executing action.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TimeoutCause {
+ /// The action's `Action.timeout` passed, plus the grace a live worker
+ /// has to report its own result.
+ ActionTimeout { timeout: Duration, grace: Duration },
+ /// The worker sent no update on the action for this long.
+ NoWorkerUpdate(Duration),
+}
+
+impl core::fmt::Display for TimeoutCause {
+ fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+ match self {
+ Self::ActionTimeout { timeout, grace } => write!(
+ f,
+ "exceeded its Action.timeout of {} seconds, and its worker had not reported within {} seconds after that",
+ timeout.as_secs_f32(),
+ grace.as_secs_f32(),
+ ),
+ Self::NoWorkerUpdate(ceiling) => write!(
+ f,
+ "timed out after {} seconds with no update from its worker",
+ ceiling.as_secs_f32(),
+ ),
+ }
+ }
+}
+
/// Maximum number of times an update to the database
/// can fail before giving up.
const MAX_UPDATE_RETRIES: usize = 5;
@@ -312,25 +340,23 @@ where
.err_tip(|| format!("Failed to upgrade weak reference to SimpleSchedulerStateManager in MatchingEngineActionStateResult::changed at attempt: {timeout_attempts}"))?;
// Check if worker is alive via registry before timing out.
- let should_timeout = simple_scheduler_state_manager
- .should_timeout_operation(&awaited_action)
- .await;
-
- if !should_timeout {
+ let Some(cause) = simple_scheduler_state_manager
+ .timeout_cause(&awaited_action)
+ .await
+ else {
// Worker is alive, continue waiting for updates
trace!(
operation_id = %awaited_action.operation_id(),
"Operation timeout check passed, worker is alive"
);
continue;
- }
+ };
warn!(
?awaited_action,
- "OperationId {} / {} timed out after {} seconds issuing a retry",
+ "OperationId {} / {} {cause}, issuing a retry",
awaited_action.operation_id(),
awaited_action.state().client_operation_id,
- self.no_event_action_timeout.as_secs_f32(),
);
simple_scheduler_state_manager
@@ -680,23 +706,41 @@ where
}
pub async fn should_timeout_operation(&self, awaited_action: &AwaitedAction) -> bool {
+ self.timeout_cause(awaited_action).await.is_some()
+ }
+
+ /// Why the scheduler should time out an executing action, or `None` if
+ /// it should not.
+ pub async fn timeout_cause(&self, awaited_action: &AwaitedAction) -> Option {
if !matches!(awaited_action.state().stage, ActionStage::Executing) {
- return false;
+ return None;
}
let now = (self.now_fn)().now();
// Honor the per-action `Action.timeout` from the RBE protocol as a
- // backend wall-clock deadline. Without this, the only enforcement is
- // the Bazel client's --test_timeout, which surfaces as TIMEOUT/NO
- // STATUS instead of a backend signal pointing at the worker.
+ // backend wall-clock deadline, in every liveness case below. The
+ // worker enforces the same timeout, but its clock starts when the
+ // command starts, after the inputs are fetched, which the scheduler
+ // cannot observe. Enforcing it at `Action.timeout` from assignment
+ // would pre-empt every live worker's own DEADLINE_EXCEEDED result, so
+ // allow `no_event_action_timeout` more, measured from the later of the
+ // assignment and the worker's last update: a live worker then reports
+ // first, and a wedged one (hung input fetch, unkillable child, broken
+ // external timeout, orphan) is still ended here.
let action_timeout = awaited_action.action_info().timeout;
if action_timeout > Duration::ZERO {
- let executing_started_at = awaited_action.state().last_transition_timestamp;
- if let Ok(elapsed) = now.duration_since(executing_started_at)
- && elapsed > action_timeout
+ let anchor = awaited_action
+ .state()
+ .last_transition_timestamp
+ .max(awaited_action.last_worker_updated_timestamp());
+ if let Ok(elapsed) = now.duration_since(anchor)
+ && elapsed > action_timeout.saturating_add(self.no_event_action_timeout)
{
- return true;
+ return Some(TimeoutCause::ActionTimeout {
+ timeout: action_timeout,
+ grace: self.no_event_action_timeout,
+ });
}
}
@@ -711,17 +755,23 @@ where
_ => WorkerLiveness::Stale,
};
+ let last_update = awaited_action.last_worker_updated_timestamp();
+ let silent_for = |ceiling: Duration| {
+ now.duration_since(last_update)
+ .is_ok_and(|elapsed| elapsed > ceiling)
+ .then_some(TimeoutCause::NoWorkerUpdate(ceiling))
+ };
+
match liveness {
- // Ours and heartbeating: only the stuck-but-alive ceiling applies,
- // and disabling that means no ceiling on a live worker.
+ // Ours and heartbeating: besides Action.timeout above, only the
+ // stuck-but-alive ceiling applies, and disabling that leaves an
+ // action with no Action.timeout no ceiling on a live worker.
WorkerLiveness::Alive => {
if self.max_executing_timeout > Duration::ZERO {
- let last_update = awaited_action.last_worker_updated_timestamp();
- if let Ok(elapsed) = now.duration_since(last_update) {
- return elapsed > self.max_executing_timeout;
- }
+ silent_for(self.max_executing_timeout)
+ } else {
+ None
}
- false
}
// Usually a peer's healthy worker, so worker_timeout_s must not
@@ -729,27 +779,15 @@ where
// max_action_executing_timeout_s defaults to disabled, so fall
// back to a ceiling rather than never timing out.
WorkerLiveness::Unknown => {
- let ceiling = if self.max_executing_timeout > Duration::ZERO {
- self.max_executing_timeout
+ if self.max_executing_timeout > Duration::ZERO {
+ silent_for(self.max_executing_timeout)
} else {
- ORPHANED_ACTION_TIMEOUT
- };
- let last_update = awaited_action.last_worker_updated_timestamp();
- match now.duration_since(last_update) {
- Ok(elapsed) => elapsed > ceiling,
- Err(_) => false,
+ silent_for(ORPHANED_ACTION_TIMEOUT)
}
}
// Registered here and gone quiet: ours, and it looks dead.
- WorkerLiveness::Stale => {
- let worker_should_update_before = awaited_action
- .last_worker_updated_timestamp()
- .checked_add(self.no_event_action_timeout)
- .unwrap_or(now);
-
- worker_should_update_before < now
- }
+ WorkerLiveness::Stale => silent_for(self.no_event_action_timeout),
}
}
@@ -929,28 +967,25 @@ where
// Re-check under the lock against freshly loaded state, and delegate
// rather than re-deriving the rule: the two copies had drifted.
- if !self.should_timeout_operation(&awaited_action).await {
+ let Some(cause) = self.timeout_cause(&awaited_action).await else {
trace!(
%operation_id,
worker_id = ?awaited_action.worker_id(),
"Operation no longer needs timing out, skipping"
);
return Ok(());
- }
+ };
warn!(
%operation_id,
worker_id = ?awaited_action.worker_id(),
+ %cause,
"Timing out operation"
);
self.assign_operation(
operation_id,
- Err(make_err!(
- Code::DeadlineExceeded,
- "Operation timed out after {} seconds",
- self.no_event_action_timeout.as_secs_f32(),
- )),
+ Err(make_err!(Code::DeadlineExceeded, "Operation {cause}")),
)
.await
}
diff --git a/nativelink-scheduler/tests/simple_scheduler_state_manager_test.rs b/nativelink-scheduler/tests/simple_scheduler_state_manager_test.rs
index eb96633b6cd..371487a668b 100644
--- a/nativelink-scheduler/tests/simple_scheduler_state_manager_test.rs
+++ b/nativelink-scheduler/tests/simple_scheduler_state_manager_test.rs
@@ -19,7 +19,9 @@ use nativelink_scheduler::awaited_action_db::{
};
use nativelink_scheduler::default_scheduler_factory::memory_awaited_action_db_factory;
use nativelink_scheduler::simple_scheduler::SimpleScheduler;
-use nativelink_scheduler::simple_scheduler_state_manager::SimpleSchedulerStateManager;
+use nativelink_scheduler::simple_scheduler_state_manager::{
+ SimpleSchedulerStateManager, TimeoutCause,
+};
use nativelink_scheduler::worker::Worker;
use nativelink_scheduler::worker_registry::WorkerRegistry;
use nativelink_scheduler::worker_scheduler::WorkerScheduler;
@@ -262,6 +264,231 @@ async fn does_not_time_out_a_live_worker_mid_action() -> Result<(), Error> {
Ok(())
}
+/// `Action.timeout` for the deadline tests below. The grace is
+/// `WORKER_TIMEOUT`, the state managers' `no_event_action_timeout`.
+const ACTION_TIMEOUT: Duration = Duration::from_mins(1);
+
+/// An action with `ACTION_TIMEOUT`, assigned to `worker_id` and executing
+/// since `started`.
+fn action_with_timeout(worker_id: &WorkerId, started: SystemTime) -> AwaitedAction {
+ let mut info = action_info(started);
+ info.timeout = ACTION_TIMEOUT;
+ let action_digest = info.digest();
+ let operation_id = OperationId::default();
+ let mut action = AwaitedAction::new(operation_id.clone(), Arc::new(info), started);
+ action.worker_set_state(
+ Arc::new(ActionState {
+ stage: ActionStage::Executing,
+ client_operation_id: operation_id,
+ action_digest,
+ last_transition_timestamp: started,
+ }),
+ started,
+ );
+ action.set_worker_id(Some(worker_id.clone()), started);
+ action
+}
+
+const fn action_timeout_cause() -> TimeoutCause {
+ TimeoutCause::ActionTimeout {
+ timeout: ACTION_TIMEOUT,
+ grace: WORKER_TIMEOUT,
+ }
+}
+
+/// Just short of `Action.timeout` plus the grace.
+const WITHIN_GRACE: Duration =
+ Duration::from_secs(ACTION_TIMEOUT.as_secs() + WORKER_TIMEOUT.as_secs() - 1);
+
+/// Just past `Action.timeout` plus the grace.
+const PAST_GRACE: Duration =
+ Duration::from_secs(ACTION_TIMEOUT.as_secs() + WORKER_TIMEOUT.as_secs() + 1);
+
+/// A heartbeating worker enforces `Action.timeout` itself, from when the
+/// command starts, and reports a completed `DEADLINE_EXCEEDED` result. The
+/// scheduler's clock starts at assignment, earlier, so ending the action at
+/// `Action.timeout` would requeue every action that runs to its timeout onto
+/// the worker still running it, and the worker's own result would then evict
+/// it. The grace lets the live worker report first.
+#[nativelink_test]
+async fn gives_a_live_worker_the_grace_to_report_its_own_timeout() -> Result<(), Error> {
+ MockClock::set_time(Duration::from_secs(NOW_TIME));
+ let worker_id = WorkerId::from(String::from("enforces-its-own-timeout"));
+ let action = action_with_timeout(&worker_id, make_system_time(0));
+
+ let registry = Arc::new(WorkerRegistry::new());
+ registry
+ .update_worker_heartbeat(&worker_id, make_system_time(WITHIN_GRACE.as_secs()))
+ .await;
+ let state_mgr = state_manager_no_executing_ceiling(registry);
+
+ MockClock::advance(WITHIN_GRACE);
+ assert!(
+ !state_mgr.should_timeout_operation(&action).await,
+ "a live worker's action past Action.timeout but within the grace is the worker's to end"
+ );
+ Ok(())
+}
+
+/// A heartbeating worker that never reports (a hung input fetch, an
+/// unkillable child, a broken `timeout_handled_externally` wrapper) is timed
+/// out by `Action.timeout` once the grace has passed, even with
+/// `max_action_executing_timeout_s` disabled.
+#[nativelink_test]
+async fn times_out_a_live_but_wedged_worker_past_the_grace() -> Result<(), Error> {
+ MockClock::set_time(Duration::from_secs(NOW_TIME));
+ let worker_id = WorkerId::from(String::from("heartbeats-but-never-reports"));
+ let action = action_with_timeout(&worker_id, make_system_time(0));
+
+ let registry = Arc::new(WorkerRegistry::new());
+ registry
+ .update_worker_heartbeat(&worker_id, make_system_time(PAST_GRACE.as_secs()))
+ .await;
+ let state_mgr = state_manager_no_executing_ceiling(registry);
+
+ MockClock::advance(PAST_GRACE);
+ let cause = state_mgr.timeout_cause(&action).await;
+ assert_eq!(
+ cause,
+ Some(action_timeout_cause()),
+ "a live worker's action past Action.timeout plus the grace must time out"
+ );
+ assert!(
+ cause
+ .unwrap()
+ .to_string()
+ .contains("Action.timeout of 60 seconds"),
+ "the client's error must name the deadline that fired"
+ );
+ Ok(())
+}
+
+/// The grace runs from the worker's last update when that is later than the
+/// assignment, so it only has to cover the time since the worker last showed
+/// signs of life.
+#[nativelink_test]
+async fn measures_the_grace_from_the_workers_last_update() -> Result<(), Error> {
+ MockClock::set_time(Duration::from_secs(NOW_TIME));
+ let worker_id = WorkerId::from(String::from("updated-after-assignment"));
+ let mut action = action_with_timeout(&worker_id, make_system_time(0));
+ let updated = Duration::from_secs(30);
+ let state = action.state().clone();
+ action.worker_set_state(state, make_system_time(updated.as_secs()));
+
+ let registry = Arc::new(WorkerRegistry::new());
+ registry
+ .update_worker_heartbeat(
+ &worker_id,
+ make_system_time((PAST_GRACE + updated).as_secs()),
+ )
+ .await;
+ let state_mgr = state_manager_no_executing_ceiling(registry);
+
+ MockClock::advance(PAST_GRACE);
+ assert!(
+ !state_mgr.should_timeout_operation(&action).await,
+ "the grace must run from the last worker update, not the assignment"
+ );
+
+ MockClock::advance(updated);
+ assert_eq!(
+ state_mgr.timeout_cause(&action).await,
+ Some(action_timeout_cause()),
+ "past Action.timeout plus the grace from the last update, the action must time out"
+ );
+ Ok(())
+}
+
+/// A peer instance's live worker enforces `Action.timeout` itself too, so it
+/// gets the same grace here.
+#[nativelink_test]
+async fn gives_a_peer_instances_worker_the_grace() -> Result<(), Error> {
+ MockClock::set_time(Duration::from_secs(NOW_TIME));
+ // Deliberately not registered here: it belongs to another instance.
+ let action = action_with_timeout(
+ &WorkerId::from(String::from("owned-by-another-scheduler")),
+ make_system_time(0),
+ );
+ let state_mgr = state_manager_no_executing_ceiling(Arc::new(WorkerRegistry::new()));
+
+ MockClock::advance(WITHIN_GRACE);
+ assert!(
+ !state_mgr.should_timeout_operation(&action).await,
+ "a peer's worker's action within the grace is the worker's to end"
+ );
+ Ok(())
+}
+
+/// An orphan nobody reaps, or a peer's worker that never reports, is timed
+/// out by `Action.timeout` past the grace rather than waiting for the orphan
+/// ceiling.
+#[nativelink_test]
+async fn times_out_an_orphan_on_action_timeout_past_the_grace() -> Result<(), Error> {
+ MockClock::set_time(Duration::from_secs(NOW_TIME));
+ let action = action_with_timeout(
+ &WorkerId::from(String::from("owner-was-scaled-down")),
+ make_system_time(0),
+ );
+ let state_mgr = state_manager_no_executing_ceiling(Arc::new(WorkerRegistry::new()));
+
+ MockClock::advance(PAST_GRACE);
+ assert_eq!(
+ state_mgr.timeout_cause(&action).await,
+ Some(action_timeout_cause()),
+ "an orphan past Action.timeout plus the grace must time out, not wait an hour"
+ );
+ Ok(())
+}
+
+/// Registry heartbeats refresh only on keepalives, so a worker whose
+/// keepalives lag reads Stale while its action is still running and still
+/// updating. `Action.timeout` must not pre-empt it there either.
+#[nativelink_test]
+async fn gives_a_stale_worker_with_a_recent_update_the_grace() -> Result<(), Error> {
+ MockClock::set_time(Duration::from_secs(NOW_TIME));
+ let worker_id = WorkerId::from(String::from("keepalives-lag"));
+ let assigned = WORKER_TIMEOUT + Duration::from_secs(30);
+ let action = action_with_timeout(&worker_id, make_system_time(assigned.as_secs()));
+
+ let registry = Arc::new(WorkerRegistry::new());
+ registry
+ .register_worker(&worker_id, make_system_time(0))
+ .await;
+ let state_mgr = state_manager_no_executing_ceiling(registry);
+
+ // Past Action.timeout, and less than worker_timeout_s since the action's
+ // last update, so only Action.timeout could fire.
+ MockClock::advance(assigned + ACTION_TIMEOUT + Duration::from_secs(1));
+ assert!(
+ !state_mgr.should_timeout_operation(&action).await,
+ "a stale worker's recently updated action must get the grace too"
+ );
+ Ok(())
+}
+
+/// A Stale worker's action past `Action.timeout` plus the grace is timed out
+/// for that reason, which is what the client is told.
+#[nativelink_test]
+async fn times_out_a_stale_worker_on_action_timeout_past_the_grace() -> Result<(), Error> {
+ MockClock::set_time(Duration::from_secs(NOW_TIME));
+ let worker_id = WorkerId::from(String::from("connected-to-us"));
+ let action = action_with_timeout(&worker_id, make_system_time(0));
+
+ let registry = Arc::new(WorkerRegistry::new());
+ registry
+ .register_worker(&worker_id, make_system_time(0))
+ .await;
+ let state_mgr = state_manager_no_executing_ceiling(registry);
+
+ MockClock::advance(PAST_GRACE);
+ assert_eq!(
+ state_mgr.timeout_cause(&action).await,
+ Some(action_timeout_cause()),
+ "a stale worker's action past Action.timeout plus the grace must name Action.timeout"
+ );
+ Ok(())
+}
+
/// A heartbeating worker stuck on one action past the executing ceiling must
/// still be timed out, which is what `max_action_executing_timeout_s` is for.
#[nativelink_test]
diff --git a/nativelink-scheduler/tests/simple_scheduler_test.rs b/nativelink-scheduler/tests/simple_scheduler_test.rs
index 351474edae6..2df93f1625e 100644
--- a/nativelink-scheduler/tests/simple_scheduler_test.rs
+++ b/nativelink-scheduler/tests/simple_scheduler_test.rs
@@ -2571,7 +2571,9 @@ async fn worker_disconnect_loop_caps_at_max_worker_loss_retries_test() -> Result
#[nativelink_test]
async fn action_timeout_is_enforced_backend_side_test() -> Result<(), Error> {
use nativelink_scheduler::awaited_action_db::AwaitedAction;
- use nativelink_scheduler::simple_scheduler_state_manager::SimpleSchedulerStateManager;
+ use nativelink_scheduler::simple_scheduler_state_manager::{
+ SimpleSchedulerStateManager, TimeoutCause,
+ };
// Anchor MockClock so MockInstantWrapped::now() == make_system_time(0).
MockClock::set_time(Duration::from_secs(NOW_TIME));
@@ -2614,12 +2616,25 @@ async fn action_timeout_is_enforced_backend_side_test() -> Result<(), Error> {
"Should not time out before Action.timeout elapses",
);
- // Advance past the 2s per-action deadline.
+ // Past the 2s per-action deadline, but within the grace a live worker
+ // has to report its own DEADLINE_EXCEEDED result.
MockClock::advance(Duration::from_secs(5));
assert!(
- state_mgr.should_timeout_operation(&awaited_action).await,
- "Scheduler must mark Executing action timed out once Action.timeout has elapsed",
+ !state_mgr.should_timeout_operation(&awaited_action).await,
+ "Should not time out within the grace after Action.timeout",
+ );
+
+ // Past Action.timeout plus the grace (no_event_action_timeout).
+ MockClock::advance(Duration::from_mins(1));
+
+ assert_eq!(
+ state_mgr.timeout_cause(&awaited_action).await,
+ Some(TimeoutCause::ActionTimeout {
+ timeout: Duration::from_secs(2),
+ grace: Duration::from_mins(1),
+ }),
+ "Scheduler must time out an Executing action once Action.timeout plus the grace has elapsed",
);
Ok(())
diff --git a/nativelink-service/Cargo.toml b/nativelink-service/Cargo.toml
index e7b5f799ebb..9910222b285 100644
--- a/nativelink-service/Cargo.toml
+++ b/nativelink-service/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-service"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-config = { path = "../nativelink-config" }
diff --git a/nativelink-service/src/worker_api_server.rs b/nativelink-service/src/worker_api_server.rs
index 86239a37256..68b84dde2d4 100644
--- a/nativelink-service/src/worker_api_server.rs
+++ b/nativelink-service/src/worker_api_server.rs
@@ -303,19 +303,38 @@ impl WorkerConnection {
};
background_spawn!("worker_api", async move {
- let mut had_going_away = false;
- while let Some(maybe_update) = connection.next().await {
- let update = match maybe_update.map(|u| u.update) {
- Ok(Some(update)) => update,
- Ok(None) => {
- tracing::warn!(worker_id=?instance.worker_id, "Empty update");
- continue;
- }
- Err(err) => {
- tracing::warn!(worker_id=?instance.worker_id, ?err, "Error from worker");
+ // The stream is read by a task of its own that only hands each
+ // message on; they are processed here, in order. Processing one
+ // inline in the reader (a result's store write, say) stopped the
+ // reads for its duration, and a worker that filled the stream's
+ // flow-control window in the meantime had nothing it was allowed
+ // to put on the wire: its keepalives waited behind the stall
+ // until this scheduler evicted it. The queue is unbounded on
+ // purpose: a worker's messages are bounded by its in-flight
+ // actions and its keepalive interval, and backpressure here is
+ // exactly the stall.
+ let (tx, mut rx) = mpsc::unbounded_channel();
+ let reader_worker_id = instance.worker_id.clone();
+ background_spawn!("worker_api_reader", async move {
+ while let Some(maybe_update) = connection.next().await {
+ let update = match maybe_update.map(|u| u.update) {
+ Ok(Some(update)) => update,
+ Ok(None) => {
+ tracing::warn!(worker_id=?reader_worker_id, "Empty update");
+ continue;
+ }
+ Err(err) => {
+ tracing::warn!(worker_id=?reader_worker_id, ?err, "Error from worker");
+ break;
+ }
+ };
+ if tx.send(update).is_err() {
break;
}
- };
+ }
+ });
+ let mut had_going_away = false;
+ while let Some(update) = rx.recv().await {
let result = match update {
Update::ConnectWorkerRequest(_connect_worker_request) => Err(make_err!(
Code::Internal,
diff --git a/nativelink-service/tests/worker_api_server_test.rs b/nativelink-service/tests/worker_api_server_test.rs
index 4b63d264380..f50c7ac28d9 100644
--- a/nativelink-service/tests/worker_api_server_test.rs
+++ b/nativelink-service/tests/worker_api_server_test.rs
@@ -1176,3 +1176,70 @@ pub async fn a_full_channel_defers_the_kill_instead_of_evicting_test()
);
Ok(())
}
+
+/// A message the scheduler is slow to process must not stop it reading the
+/// worker's stream: the keepalives behind it still come off the wire (here,
+/// a one-slot channel the test could not fill otherwise), and are taken in,
+/// in order, once the slow one is done. Reading inline, the stream's
+/// flow-control window filled behind a slow result and the worker's
+/// keepalives never left it, so this scheduler evicted a worker that was
+/// only waiting for it.
+#[nativelink_test]
+pub async fn a_slow_result_does_not_stop_the_stream_reads_test()
+-> Result<(), Box> {
+ let mut test_context = setup_api_server(BASE_WORKER_TIMEOUT_S, Box::new(static_now_fn)).await?;
+ let operation_id = dispatch(&test_context, 1).await?;
+ let update_for_worker = test_context
+ .connection_worker_stream
+ .next()
+ .await
+ .expect("Worker stream ended early")?
+ .update
+ .expect("Expected update field to be populated");
+ let update_for_worker::Update::StartAction(start_execute) = update_for_worker else {
+ panic!("Expected StartAction message");
+ };
+ assert_eq!(operation_id.to_string(), start_execute.operation_id);
+
+ // The result's store write is not answered yet: processing it blocks.
+ test_context
+ .worker_stream
+ .send(Update::ExecuteResult(ExecuteResult {
+ instance_name: "instance_name".to_string(),
+ operation_id: operation_id.to_string(),
+ result: Some(execute_result::Result::InternalError(
+ make_err!(tonic::Code::Internal, "the action failed").into(),
+ )),
+ resource_usage: None,
+ }))
+ .await?;
+
+ // The keepalives behind it are still read off the stream.
+ for free_memory_kb in 1..=8u64 {
+ tokio::time::timeout(
+ Duration::from_secs(5),
+ test_context
+ .worker_stream
+ .send(Update::KeepAliveRequest(KeepAliveRequest {
+ load: Some(WorkerLoad { free_memory_kb }),
+ })),
+ )
+ .await
+ .map_err(|_| {
+ make_err!(
+ tonic::Code::DeadlineExceeded,
+ "the stream read stalled behind the result"
+ )
+ })??;
+ }
+
+ // Once the result is through, the keepalives are taken in after it.
+ let (operation_id_seen, worker_id, _) = test_context
+ .state_manager
+ .expect_update_operation(Ok(()))
+ .await;
+ assert_eq!(operation_id_seen, operation_id);
+ assert_eq!(worker_id, test_context.worker_id);
+ summary_after_report(&test_context, 8).await;
+ Ok(())
+}
diff --git a/nativelink-store/Cargo.toml b/nativelink-store/Cargo.toml
index 5cfee9301f2..6699ea8df8f 100644
--- a/nativelink-store/Cargo.toml
+++ b/nativelink-store/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-store"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-config = { path = "../nativelink-config" }
diff --git a/nativelink-store/src/fast_slow_store.rs b/nativelink-store/src/fast_slow_store.rs
index 299d432552f..746858c5593 100644
--- a/nativelink-store/src/fast_slow_store.rs
+++ b/nativelink-store/src/fast_slow_store.rs
@@ -276,6 +276,19 @@ impl FastSlowStore {
offset: u64,
length: Option,
) -> Result<(), Error> {
+ // A `NotFound` for a digest key carries the digest as typed
+ // context, so the worker can report it as the missing input and the
+ // client re-uploads exactly that blob; see `ErrorContext`.
+ fn name_missing_digest(err: Error, key: &StoreKey<'_>) -> Error {
+ if let StoreKey::Digest(d) = key {
+ err.with_context(ErrorContext::MissingDigest {
+ hash: d.packed_hash().to_string(),
+ size: d.size_bytes().try_into().unwrap_or(i64::MAX),
+ })
+ } else {
+ err
+ }
+ }
let reader_stream_size = if self
.slow_store
.inner_store(Some(key.borrow()))
@@ -294,20 +307,15 @@ impl FastSlowStore {
.await
.err_tip(|| "Failed to run has() on slow store")?
.ok_or_else(|| {
- let err = make_err!(
- Code::NotFound,
- "Object {} not found in either fast or slow store. \
- If using multiple workers, ensure all workers share the same CAS storage path.",
- key.as_str()
- );
- if let StoreKey::Digest(d) = key.borrow() {
- err.with_context(ErrorContext::MissingDigest {
- hash: d.packed_hash().to_string(),
- size: d.size_bytes().try_into().unwrap_or(i64::MAX),
- })
- } else {
- err
- }
+ name_missing_digest(
+ make_err!(
+ Code::NotFound,
+ "Object {} not found in either fast or slow store. \
+ If using multiple workers, ensure all workers share the same CAS storage path.",
+ key.as_str()
+ ),
+ &key,
+ )
})?
)
};
@@ -387,7 +395,12 @@ impl FastSlowStore {
)
}
Err(err) => match slow_res {
- Err(slow_err) if slow_err.code == Code::NotFound => Err(slow_err),
+ // The slow store lost the blob under the read (an evicting
+ // tier behind a gRPC store, say): the same missing input as
+ // a blob that was never there.
+ Err(slow_err) if slow_err.code == Code::NotFound => {
+ Err(name_missing_digest(slow_err, &key))
+ }
_ => fast_res.merge(slow_res).merge(Err(err)),
},
}
@@ -811,9 +824,16 @@ impl StoreDriver for FastSlowStore {
) -> Result<(), Error> {
// `has()` can report a stale map entry whose file is gone, so
// get_part may still return NotFound; fall through to the slow
- // store unless we have already streamed bytes to the caller.
+ // store. With bytes already streamed to the caller that is a resume:
+ // a digest names an immutable blob, so what the fast tier sent is a
+ // correct prefix and the slow store serves the rest from there (a
+ // Redis tier evicting the key under a read is the live case). Any
+ // other key could hold different content in each tier, so a short
+ // read there stays an error.
// One existence check, reused: this is the hot read path and `has()`
// on a filesystem fast store is a syscall.
+ let mut offset = offset;
+ let mut length = length;
let in_fast_store = self.fast_store.has(key.borrow()).await?.is_some();
if !in_fast_store {
record_store_tier_read("fast", "miss");
@@ -846,6 +866,28 @@ impl StoreDriver for FastSlowStore {
warn!(%key, ?e, "Stale fast-store map entry; falling through to slow store");
// fall through to populate path
}
+ Err(e) if e.code == Code::NotFound && matches!(key, StoreKey::Digest(_)) => {
+ let sent = writer.get_bytes_written() - bytes_before;
+ if let Some(wanted) = length {
+ let rest = wanted.saturating_sub(sent);
+ if rest == 0 {
+ return Ok(());
+ }
+ length = Some(rest);
+ }
+ offset += sent;
+ self.metrics
+ .fast_store_short_reads_resumed
+ .fetch_add(1, Ordering::Acquire);
+ record_store_tier_read("fast", "short");
+ warn!(
+ %key,
+ sent,
+ ?e,
+ "Fast store ended short; resuming from the slow store"
+ );
+ // fall through to populate path, from the next byte
+ }
Err(e) => return Err(e),
}
}
@@ -1009,6 +1051,10 @@ struct FastSlowStoreMetrics {
huge_blob_dedup_bypasses: AtomicU64,
#[metric(help = "Stale fast-store map entries that fell through to the slow store")]
fast_store_stale_map_falls_through: AtomicU64,
+ #[metric(
+ help = "Reads of a digest the fast store ended short, resumed from the slow store at the byte it stopped"
+ )]
+ fast_store_short_reads_resumed: AtomicU64,
}
/// Maximum time a follower will wait on the leader-populator before
diff --git a/nativelink-store/src/filesystem_store.rs b/nativelink-store/src/filesystem_store.rs
index 51459ca2acd..dfb7e0d099e 100644
--- a/nativelink-store/src/filesystem_store.rs
+++ b/nativelink-store/src/filesystem_store.rs
@@ -1518,7 +1518,7 @@ impl FilesystemStore {
fn get_and_update_generation(&self) -> Result {
self.next_generation
- .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |generation| {
+ .try_update(Ordering::Relaxed, Ordering::Relaxed, |generation| {
generation.checked_add(1)
})
.map(Generation::new)
diff --git a/nativelink-store/tests/fast_slow_store_test.rs b/nativelink-store/tests/fast_slow_store_test.rs
index d234f2ac804..e3f6c11f82a 100644
--- a/nativelink-store/tests/fast_slow_store_test.rs
+++ b/nativelink-store/tests/fast_slow_store_test.rs
@@ -23,7 +23,7 @@ use futures::future::join_all;
use nativelink_config::stores::{
FastSlowSpec, FilesystemSpec, MemorySpec, NoopSpec, StoreDirection, StoreSpec,
};
-use nativelink_error::{Code, Error, ResultExt, make_err};
+use nativelink_error::{Code, Error, ErrorContext, ResultExt, make_err};
use nativelink_macro::nativelink_test;
use nativelink_metric::MetricsComponent;
use nativelink_store::fast_slow_store::FastSlowStore;
@@ -2019,9 +2019,12 @@ async fn get_part_falls_through_to_slow_on_stale_fast_map_entry() -> Result<(),
Ok(())
}
-/// `NotFound` after partial bytes must propagate, not retry (would corrupt).
+/// `NotFound` after partial bytes of a digest is a resume: the bytes the
+/// fast store sent are a correct prefix of an immutable blob, so the slow
+/// store serves the rest from where the fast one stopped and the caller
+/// sees the whole blob.
#[nativelink_test]
-async fn get_part_propagates_not_found_after_partial_fast_read() -> Result<(), Error> {
+async fn get_part_resumes_from_slow_after_partial_fast_read() -> Result<(), Error> {
let original_data = make_random_data(MEGABYTE_SZ);
let digest = DigestInfo::try_new(VALID_HASH, original_data.len()).unwrap();
let (fast_slow_store, slow_store, fast_get_part_calls) =
@@ -2031,19 +2034,156 @@ async fn get_part_propagates_not_found_after_partial_fast_read() -> Result<(), E
.update_oneshot(digest, original_data.clone().into())
.await?;
- let result = fast_slow_store.get_part_unchunked(digest, 0, None).await;
+ let served = fast_slow_store
+ .get_part_unchunked(digest, 0, None)
+ .await
+ .err_tip(|| "a short fast read of a digest must resume from the slow store")?;
- let err = result.expect_err("partial fast read then NotFound must not fall through");
- assert_eq!(
- err.code,
- Code::NotFound,
- "original NotFound must propagate, got: {err:?}"
- );
+ // The stale fast store sends sixteen zero bytes and stops; the slow
+ // store has to have served everything after them.
+ assert_eq!(served.len(), original_data.len());
+ assert_eq!(&served[16..], &original_data[16..]);
assert_eq!(
fast_get_part_calls.load(Ordering::Acquire),
1,
"fast store get_part must be attempted exactly once"
);
+ // A range read resumes within the range.
+ let served = fast_slow_store
+ .get_part_unchunked(digest, 100, Some(1000))
+ .await?;
+ assert_eq!(served.len(), 1000);
+ assert_eq!(&served[16..], &original_data[116..1100]);
+
+ Ok(())
+}
+
+/// A key that is not a digest could hold different content in each tier,
+/// so a short read of one stays an error rather than splicing the tiers.
+#[nativelink_test]
+async fn get_part_propagates_not_found_after_partial_fast_read_of_a_plain_key() -> Result<(), Error>
+{
+ let original_data = make_random_data(MEGABYTE_SZ);
+ let key = StoreKey::new_str("some-action-cache-entry");
+ let (fast_slow_store, slow_store, fast_get_part_calls) =
+ make_stores_with_stale_fast(original_data.len() as u64, 16);
+
+ slow_store
+ .update_oneshot(key.borrow(), original_data.clone().into())
+ .await?;
+
+ let err = fast_slow_store
+ .get_part_unchunked(key.borrow(), 0, None)
+ .await
+ .expect_err("a short fast read of a plain key must not resume");
+ assert_eq!(err.code, Code::NotFound, "{err:?}");
+ assert_eq!(fast_get_part_calls.load(Ordering::Acquire), 1);
+
+ Ok(())
+}
+
+/// A slow store that sends part of a blob and then loses it, the way an
+/// evicting tier behind a gRPC store does.
+#[derive(MetricsComponent)]
+struct PartialThenGoneStore {
+ bytes_before_error: usize,
+}
+
+#[async_trait]
+impl StoreDriver for PartialThenGoneStore {
+ async fn post_init(self: Arc) -> Result<(), Error> {
+ Ok(())
+ }
+
+ async fn has_with_results(
+ self: Pin<&Self>,
+ keys: &[StoreKey<'_>],
+ results: &mut [Option],
+ ) -> Result<(), Error> {
+ for (key, result) in keys.iter().zip(results.iter_mut()) {
+ *result = match key {
+ StoreKey::Digest(d) => Some(d.size_bytes()),
+ StoreKey::Str(_) => None,
+ };
+ }
+ Ok(())
+ }
+
+ async fn update(
+ self: Pin<&Self>,
+ _key: StoreKey<'_>,
+ mut reader: DropCloserReadHalf,
+ _size_info: UploadSizeInfo,
+ ) -> Result {
+ reader.drain().await
+ }
+
+ async fn get_part(
+ self: Pin<&Self>,
+ _key: StoreKey<'_>,
+ writer: &mut DropCloserWriteHalf,
+ _offset: u64,
+ _length: Option,
+ ) -> Result<(), Error> {
+ writer
+ .send(Bytes::from(vec![1u8; self.bytes_before_error]))
+ .await?;
+ Err(make_err!(Code::NotFound, "the key is gone under the read"))
+ }
+
+ fn inner_store(&self, _digest: Option) -> &'_ dyn StoreDriver {
+ self
+ }
+
+ fn as_any(&self) -> &(dyn core::any::Any + Sync + Send + 'static) {
+ self
+ }
+
+ fn as_any_arc(self: Arc) -> Arc {
+ self
+ }
+
+ fn register_remove_callback(self: Arc, _callback: RemoveCallback) -> Result<(), Error> {
+ Ok(())
+ }
+}
+
+default_health_status_indicator!(PartialThenGoneStore);
+
+/// A `NotFound` the slow store raises under the read names the digest, the
+/// way one for a blob that was never there does, so the worker reports the
+/// missing input and the client re-uploads exactly that blob.
+#[nativelink_test]
+async fn slow_store_not_found_under_the_read_names_the_digest() -> Result<(), Error> {
+ let digest = DigestInfo::try_new(VALID_HASH, 4096).unwrap();
+ let fast_slow_store = Store::new(FastSlowStore::new(
+ &FastSlowSpec {
+ fast: StoreSpec::Memory(MemorySpec::default()),
+ slow: StoreSpec::Memory(MemorySpec::default()),
+ fast_direction: StoreDirection::default(),
+ slow_direction: StoreDirection::default(),
+ bypass_dedup_threshold_bytes: 0,
+ },
+ Store::new(MemoryStore::new(&MemorySpec::default())),
+ Store::new(Arc::new(PartialThenGoneStore {
+ bytes_before_error: 1024,
+ })),
+ ));
+
+ let err = fast_slow_store
+ .get_part_unchunked(digest, 0, None)
+ .await
+ .expect_err("the slow store lost the blob under the read");
+ assert_eq!(err.code, Code::NotFound, "{err:?}");
+ assert_eq!(
+ err.context,
+ ErrorContext::MissingDigest {
+ hash: digest.packed_hash().to_string(),
+ size: 4096,
+ },
+ "{err:?}"
+ );
+
Ok(())
}
diff --git a/nativelink-test/fuzz/Cargo.lock b/nativelink-test/fuzz/Cargo.lock
index f1b80903e7b..b5da392017e 100644
--- a/nativelink-test/fuzz/Cargo.lock
+++ b/nativelink-test/fuzz/Cargo.lock
@@ -1022,7 +1022,7 @@ dependencies = [
[[package]]
name = "nativelink-config"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"byte-unit",
"humantime",
@@ -1037,7 +1037,7 @@ dependencies = [
[[package]]
name = "nativelink-error"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"base64",
"mongodb",
@@ -1069,7 +1069,7 @@ dependencies = [
[[package]]
name = "nativelink-metric"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"async-lock",
"nativelink-metric-macro-derive",
@@ -1080,7 +1080,7 @@ dependencies = [
[[package]]
name = "nativelink-metric-macro-derive"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"proc-macro2",
"quote",
@@ -1089,7 +1089,7 @@ dependencies = [
[[package]]
name = "nativelink-proto"
-version = "1.7.2"
+version = "1.7.3"
dependencies = [
"derive_more",
"prost",
diff --git a/nativelink-util/Cargo.toml b/nativelink-util/Cargo.toml
index af810191aef..90a34b86a71 100644
--- a/nativelink-util/Cargo.toml
+++ b/nativelink-util/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-util"
-version = "1.7.2"
+version = "1.7.3"
[dependencies]
nativelink-config = { path = "../nativelink-config" }
@@ -105,7 +105,7 @@ opentelemetry-proto = { version = "0.32.0", default-features = false, features =
"gen-tonic",
"metrics",
] }
-serial_test = { version = "3.2.0", features = [
+serial_test = { version = "4.0.0", features = [
"async",
], default-features = false }
diff --git a/nativelink-worker/Cargo.toml b/nativelink-worker/Cargo.toml
index 58184fc0791..01cef3d9e9f 100644
--- a/nativelink-worker/Cargo.toml
+++ b/nativelink-worker/Cargo.toml
@@ -4,7 +4,7 @@ lints.workspace = true
[package]
edition = "2024"
name = "nativelink-worker"
-version = "1.7.2"
+version = "1.7.3"
[features]
nix = []
@@ -74,7 +74,7 @@ pretty_assertions = { version = "1.4.1", features = [
"std",
], default-features = false }
prost-types = { version = "0.14.4", default-features = false }
-serial_test = { version = "3.2.0", features = [
+serial_test = { version = "4.0.0", features = [
"async",
], default-features = false }
tempfile = { version = "3.15.0", default-features = false }
diff --git a/nativelink-worker/src/local_worker.rs b/nativelink-worker/src/local_worker.rs
index 0f93d6b86b3..79d1be494c1 100644
--- a/nativelink-worker/src/local_worker.rs
+++ b/nativelink-worker/src/local_worker.rs
@@ -110,6 +110,49 @@ fn reconnect_delay(attempt: u32) -> Duration {
/// `cas_server.rs` must also be updated.
const DEFAULT_ENDPOINT_TIMEOUT_S: f32 = 5.;
+/// Keepalive defaults for the scheduler connection, the same as a store's
+/// gRPC endpoint (`tls_utils::endpoint`).
+const DEFAULT_TCP_KEEPALIVE: Duration = Duration::from_secs(30);
+const DEFAULT_HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
+const DEFAULT_HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
+
+/// A configured number of seconds, or the default when unset (0).
+const fn seconds_or(configured: u64, default: Duration) -> Duration {
+ if configured > 0 {
+ Duration::from_secs(configured)
+ } else {
+ default
+ }
+}
+
+/// Sends `GoingAway` within `deadline`. Logged rather than fatal: the stream
+/// close that follows is what removes the worker, this only says why.
+async fn send_going_away(
+ grpc_client: &mut T,
+ drain: bool,
+ deadline: Duration,
+) -> Result<(), Error> {
+ match time::timeout(deadline, grpc_client.going_away(GoingAwayRequest { drain })).await {
+ Ok(Ok(())) => Ok(()),
+ Ok(Err(e)) => {
+ error!(?e, drain, "Failed to send GoingAwayRequest");
+ Err(e.append("Failed to send GoingAwayRequest"))
+ }
+ Err(_) => {
+ error!(
+ drain,
+ deadline_s = deadline.as_secs_f32(),
+ "GoingAwayRequest did not complete; the scheduler connection is stalled"
+ );
+ Err(make_err!(
+ Code::DeadlineExceeded,
+ "GoingAwayRequest did not complete within {}s",
+ deadline.as_secs_f32()
+ ))
+ }
+ }
+}
+
/// Default maximum amount of time a task is allowed to run for.
/// If this value gets modified the documentation in `cas_server.rs` must also be updated.
const DEFAULT_MAX_ACTION_TIMEOUT: Duration = Duration::from_mins(20);
@@ -139,7 +182,7 @@ struct LocalWorkerImpl<'a, T: WorkerApiClientTrait + 'static, U: RunningActionsM
// always be zero if there are no actions running and no actions being waited
// on by the scheduler.
actions_in_transit: Arc,
- accepted_action: AtomicBool,
+ accepted_action: Arc,
/// The scheduler said it understands `ExecuteAccepted` and
/// `ExecuteDeclined`. Without it the worker runs whatever it is sent,
/// as every earlier release did.
@@ -294,7 +337,7 @@ impl<'a, T: WorkerApiClientTrait + 'static, U: RunningActionsManager> LocalWorke
// always be zero if there are no actions running and no actions being waited
// on by the scheduler.
actions_in_transit: Arc::new(AtomicU64::new(0)),
- accepted_action: AtomicBool::new(false),
+ accepted_action: Arc::new(AtomicBool::new(false)),
metrics,
}
}
@@ -319,13 +362,47 @@ impl<'a, T: WorkerApiClientTrait + 'static, U: RunningActionsManager> LocalWorke
/// Tells the scheduler the action is refused; only meaningful when it
/// understands the message.
- async fn decline(&self, operation_id: String, refusal: Refusal) -> Result<(), Error> {
+ /// The one message bounded by the endpoint timeout: `GoingAway` at
+ /// shutdown, so a stalled connection cannot hold a pod past its
+ /// termination grace. Everything else the worker sends rides on the
+ /// `ConnectWorker` stream unbounded, and the scheduler's liveness
+ /// timeout is the clock that evicts a worker it cannot hear.
+ fn going_away_deadline(&self) -> Duration {
+ Duration::from_secs_f32(
+ self.config
+ .worker_api_endpoint
+ .timeout
+ .unwrap_or(DEFAULT_ENDPOINT_TIMEOUT_S),
+ )
+ }
+
+ /// A message to the scheduler the run loop does not wait for inline:
+ /// it rides in the loop's future set, so the loop keeps reading its
+ /// stream while the message waits its turn on the one-slot channel
+ /// feeding the `ConnectWorker` stream. No deadline on it: a message
+ /// waits behind whatever the scheduler is still processing, and a
+ /// scheduler that stops hearing this worker closes the stream, which
+ /// the loop now sees.
+ fn scheduler_send(
+ &self,
+ what: &'static str,
+ send: impl Future