From 5dd457928808cfda5e725cd68ac614f1bc54876f Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 00:26:20 +0000 Subject: [PATCH 1/9] server: bound in-flight CAS ByteStream writes, shed excess with UNAVAILABLE Each in-flight CAS Write past the Contains check holds a 1MB chunk buffer, zstd output and pipe/gRPC buffers before anything reaches disk. One tenant's upload burst drove bazel-l1-us-west-9 to 8k concurrent writes and a 54GB heap against GOMEMLIMIT=40GiB, stalling the process (health polls failed, 130k CLOSE-WAIT sockets). BAZEL_REMOTE_MAX_INFLIGHT_CAS_WRITES=N caps writes that reach the Put path; excess writes are answered UNAVAILABLE immediately (logged WRITE SHED), which the FA shim already treats as a dropped write rather than a failed build. Reads, AC, FindMissingBlobs and the existing-blob early return are untouched. Unset = unlimited (current behaviour). Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- main.go | 9 +++++++++ server/grpc.go | 8 +++++++- server/grpc_bytestream.go | 11 +++++++++++ server/runtime.go | 32 ++++++++++++++++++++++++++++++++ 4 files changed, 59 insertions(+), 1 deletion(-) diff --git a/main.go b/main.go index 3c6b8f6..230a879 100644 --- a/main.go +++ b/main.go @@ -10,6 +10,7 @@ import ( "os" "os/signal" "runtime" + "strconv" "strings" "syscall" "time" @@ -610,6 +611,14 @@ func startGrpcServer(c *config.Config, grpcServer **grpc.Server, log.Println("Starting gRPC server on address", addr) var grpcServerOpts []server.GRPCServerOption + if v := os.Getenv("BAZEL_REMOTE_MAX_INFLIGHT_CAS_WRITES"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n < 0 { + return fmt.Errorf("invalid BAZEL_REMOTE_MAX_INFLIGHT_CAS_WRITES %q", v) + } + log.Printf("gRPC ByteStream max in-flight CAS writes: %d", n) + grpcServerOpts = append(grpcServerOpts, server.WithMaxInflightCASWrites(n)) + } if c.ReadChunkSizeBytes > 0 { log.Printf("gRPC ByteStream read chunk size: %d bytes", c.ReadChunkSizeBytes) grpcServerOpts = append(grpcServerOpts, server.WithReadChunkSizeBytes(c.ReadChunkSizeBytes)) diff --git a/server/grpc.go b/server/grpc.go index 1d427e8..bf036e6 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -50,7 +50,13 @@ type grpcServer struct { readChunkSizeBytes int64 runtimeMetrics RuntimeMetrics readLimiter *readLimiter - sourceBuffers *sourceBufferPool + // casWriteSlots bounds in-flight CAS ByteStream Writes past the Contains + // check. Each one holds a 1MB chunk buffer, the zstd output and pipe/gRPC + // buffers before anything reaches disk, so an unbounded burst from one + // tenant (8k concurrent writes observed) takes the heap past GOMEMLIMIT + // and stalls the process. nil = unlimited. + casWriteSlots chan struct{} + sourceBuffers *sourceBufferPool // writePayloadConsumed, when set, is invoked once per fully-consumed // ByteStream WriteRequest; see WithWritePayloadConsumed. writePayloadConsumed func(*bytestream.WriteRequest) diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go index e9e8ad4..a8ab82b 100644 --- a/server/grpc_bytestream.go +++ b/server/grpc_bytestream.go @@ -428,6 +428,10 @@ func (s *grpcServer) parseWriteResource(r string) (string, int64, casblob.Compre } var errWriteOffset error = errors.New("bytestream writes from non-zero offsets are unsupported") + +// UNAVAILABLE is the connection-class code clients treat as "drop this +// write", never a failed build. +var errCASWriteShed = status.Error(codes.Unavailable, "too many in-flight CAS writes") var errDecoderPoolFail error = errors.New("failed to get DecoderWrapper from pool") func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { @@ -532,6 +536,12 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { return } + if !s.tryAcquireCASWriteSlot() { + s.accessLogger.Printf("GRPC BYTESTREAM WRITE SHED: %s", resourceName) + recvResult <- errCASWriteShed + return + } + var rc io.ReadCloser = pr if cmp == casblob.Zstandard { dec, ok := decoderPool.Get().(*syncpool.DecoderWrapper) @@ -550,6 +560,7 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { } go func() { + defer s.releaseCASWriteSlot() defer func() { _ = rc.Close() }() err := s.cache.Put(srv.Context(), cache.CAS, hash, size, rc) putResult <- err diff --git a/server/runtime.go b/server/runtime.go index fa85ac1..c0465e4 100644 --- a/server/runtime.go +++ b/server/runtime.go @@ -81,6 +81,38 @@ func WithReadChunkSizeBytes(chunkSizeBytes int64) GRPCServerOption { } } +// WithMaxInflightCASWrites caps concurrent CAS ByteStream Writes that reach +// the Put path; excess writes are shed with UNAVAILABLE. 0 = unlimited. +func WithMaxInflightCASWrites(n int) GRPCServerOption { + return func(s *grpcServer) error { + if n < 0 { + return fmt.Errorf("max inflight CAS writes must not be negative: %d", n) + } + if n > 0 { + s.casWriteSlots = make(chan struct{}, n) + } + return nil + } +} + +func (s *grpcServer) tryAcquireCASWriteSlot() bool { + if s.casWriteSlots == nil { + return true + } + select { + case s.casWriteSlots <- struct{}{}: + return true + default: + return false + } +} + +func (s *grpcServer) releaseCASWriteSlot() { + if s.casWriteSlots != nil { + <-s.casWriteSlots + } +} + // WithWritePayloadConsumed registers a callback the ByteStream Write handler // invokes exactly once per received message, after that message's Data has // been fully consumed (the pipe write to the cache reader returned). It is From 2e1bc12395b49545c3160f8b72d00fe53d1feecd Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 00:29:33 +0000 Subject: [PATCH 2/9] server: test CAS write shed when in-flight cap is reached Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- server/grpc_test.go | 72 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/server/grpc_test.go b/server/grpc_test.go index d770cd6..2e9eda2 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -3105,3 +3105,75 @@ func TestInsufficientStorageWhenProxyTriesToStoreAc(t *testing.T) { assertStatusCodeFromError(t, err, codes.ResourceExhausted) } } + +func TestGrpcByteStreamWriteShedWhenInflightCapReached(t *testing.T) { + fixture := grpcTestSetupInternal(t, false, WithMaxInflightCASWrites(1)) + defer os.RemoveAll(fixture.tempdir) + + writeResource := func(blob []byte, hash string) string { + return fmt.Sprintf("uploads/%s/blobs/%s/%d", uuid.New().String(), hash, len(blob)) + } + fullWrite := func(blob []byte, hash string) error { + bswc, err := fixture.bsClient.Write(ctx) + if err != nil { + t.Fatal(err) + } + err = bswc.Send(&bytestream.WriteRequest{ + ResourceName: writeResource(blob, hash), + FinishWrite: true, + Data: blob, + }) + if err != nil && err != io.EOF { + t.Fatal(err) + } + _, err = bswc.CloseAndRecv() + return err + } + + heldBlob, heldHash := testutils.RandomDataAndHash(64) + heldResource := writeResource(heldBlob, heldHash) + held, err := fixture.bsClient.Write(ctx) + if err != nil { + t.Fatal(err) + } + err = held.Send(&bytestream.WriteRequest{ + ResourceName: heldResource, + Data: heldBlob[:32], + }) + if err != nil { + t.Fatal(err) + } + // Let the server consume the first chunk and take the only slot. + time.Sleep(200 * time.Millisecond) + + otherBlob, otherHash := testutils.RandomDataAndHash(64) + err = fullWrite(otherBlob, otherHash) + if status.Code(err) != codes.Unavailable { + t.Fatalf("expected codes.Unavailable while a write holds the slot, got %v", err) + } + + if err := fullWrite(otherBlob, otherHash); status.Code(err) != codes.Unavailable { + t.Fatalf("expected second shed, got %v", err) + } + + err = held.Send(&bytestream.WriteRequest{ + ResourceName: heldResource, + FinishWrite: true, + Data: heldBlob[32:], + WriteOffset: 32, + }) + if err != nil && err != io.EOF { + t.Fatal(err) + } + if _, err := held.CloseAndRecv(); err != nil { + t.Fatalf("held write should complete: %v", err) + } + + if err := fullWrite(otherBlob, otherHash); err != nil { + t.Fatalf("write after slot release should succeed: %v", err) + } + // Now present: the Contains early return answers before the gate. + if err := fullWrite(otherBlob, otherHash); err != nil { + t.Fatalf("duplicate write of present blob should succeed: %v", err) + } +} From dfef1e7a72d42f804feacb71805904974042037c Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 00:31:48 +0000 Subject: [PATCH 3/9] server: log shed CAS writes once Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- server/grpc_bytestream.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go index a8ab82b..6b562ed 100644 --- a/server/grpc_bytestream.go +++ b/server/grpc_bytestream.go @@ -636,8 +636,10 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { } _ = pw.CloseWithError(err) - s.accessLogger.Printf("GRPC BYTESTREAM WRITE FAILED: %s %s", - resourceName, err.Error()) + if err != errCASWriteShed { + s.accessLogger.Printf("GRPC BYTESTREAM WRITE FAILED: %s %s", + resourceName, err.Error()) + } return err } From 6656eb09d40c92e7788eba58ba5e16e016a20b5d Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 00:34:48 +0000 Subject: [PATCH 4/9] main: optional static gRPC stream window to bound per-write buffering Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- main.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/main.go b/main.go index 230a879..5940ffa 100644 --- a/main.go +++ b/main.go @@ -499,6 +499,17 @@ func startGrpcServer(c *config.Config, grpcServer **grpc.Server, grpcSem *semaphore.Weighted, diskCache disk.Cache) error { opts := []grpc.ServerOption{} + // A static per-stream window bounds how many bytes a client can push + // ahead of the (disk-bound) Put goroutine, so memory per in-flight or + // shed write is this instead of the BDP-grown default (up to 16MB). + if v := os.Getenv("BAZEL_REMOTE_GRPC_INITIAL_WINDOW_SIZE"); v != "" { + n, err := strconv.ParseInt(v, 10, 32) + if err != nil || n < 65536 { + return fmt.Errorf("invalid BAZEL_REMOTE_GRPC_INITIAL_WINDOW_SIZE %q (bytes, >= 65536)", v) + } + log.Printf("gRPC initial stream window: %d bytes", n) + opts = append(opts, grpc.InitialWindowSize(int32(n)), grpc.InitialConnWindowSize(int32(n)*16)) + } streamInterceptors := []grpc.StreamServerInterceptor{} unaryInterceptors := []grpc.UnaryServerInterceptor{} From cb621055fe6c71922dfb8dbcaf40656ef7f3237b Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 00:46:41 +0000 Subject: [PATCH 5/9] server: acquire CAS write slot after decoder setup so error paths never leak it Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- server/grpc_bytestream.go | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go index 6b562ed..f24e1fd 100644 --- a/server/grpc_bytestream.go +++ b/server/grpc_bytestream.go @@ -536,12 +536,6 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { return } - if !s.tryAcquireCASWriteSlot() { - s.accessLogger.Printf("GRPC BYTESTREAM WRITE SHED: %s", resourceName) - recvResult <- errCASWriteShed - return - } - var rc io.ReadCloser = pr if cmp == casblob.Zstandard { dec, ok := decoderPool.Get().(*syncpool.DecoderWrapper) @@ -559,6 +553,16 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { rc = dec.IOReadCloser() } + // Acquire last so every earlier error path returns + // without holding a slot; the Put goroutine owns the + // release. + if !s.tryAcquireCASWriteSlot() { + _ = rc.Close() + s.accessLogger.Printf("GRPC BYTESTREAM WRITE SHED: %s", resourceName) + recvResult <- errCASWriteShed + return + } + go func() { defer s.releaseCASWriteSlot() defer func() { _ = rc.Close() }() From 53862ab0281f6161d930cb8f15bbd1eca0fb0c3e Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 00:58:09 +0000 Subject: [PATCH 6/9] server: acknowledge new CAS writes at the admission gate instead of returning UNAVAILABLE When the in-flight CAS write cap is reached, answer the first frame with committed_size exactly as the existing-blob path does and close the stream, so the client stops sending and never retries. The blob is not stored and is simply a future miss; bazel-remote's AC dependency check keeps an ActionResult that references it from being served as a hit. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- server/grpc_bytestream.go | 27 ++++++++++++++++++--------- server/grpc_test.go | 17 ++++++++++------- 2 files changed, 28 insertions(+), 16 deletions(-) diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go index f24e1fd..81d1e58 100644 --- a/server/grpc_bytestream.go +++ b/server/grpc_bytestream.go @@ -431,7 +431,10 @@ var errWriteOffset error = errors.New("bytestream writes from non-zero offsets a // UNAVAILABLE is the connection-class code clients treat as "drop this // write", never a failed build. -var errCASWriteShed = status.Error(codes.Unavailable, "too many in-flight CAS writes") +// errCASWriteAbsorbed signals a new CAS write that was acknowledged at the +// admission gate without being stored: the client sees committed_size as if +// the blob existed and stops sending, and the blob is simply a future miss. +var errCASWriteAbsorbed = errors.New("cas write absorbed at admission gate") var errDecoderPoolFail error = errors.New("failed to get DecoderWrapper from pool") func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { @@ -558,8 +561,12 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { // release. if !s.tryAcquireCASWriteSlot() { _ = rc.Close() - s.accessLogger.Printf("GRPC BYTESTREAM WRITE SHED: %s", resourceName) - recvResult <- errCASWriteShed + if cmp == casblob.Identity { + resp.CommittedSize = size + } else { + resp.CommittedSize = -1 + } + putResult <- errCASWriteAbsorbed return } @@ -640,10 +647,8 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { } _ = pw.CloseWithError(err) - if err != errCASWriteShed { - s.accessLogger.Printf("GRPC BYTESTREAM WRITE FAILED: %s %s", - resourceName, err.Error()) - } + s.accessLogger.Printf("GRPC BYTESTREAM WRITE FAILED: %s %s", + resourceName, err.Error()) return err } @@ -653,8 +658,12 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { default: } - if err == io.EOF { - s.accessLogger.Printf("GRPC BYTESTREAM SKIPPED WRITE: %s", resourceName) + if err == io.EOF || err == errCASWriteAbsorbed { + if err == errCASWriteAbsorbed { + s.accessLogger.Printf("GRPC BYTESTREAM WRITE ABSORBED: %s", resourceName) + } else { + s.accessLogger.Printf("GRPC BYTESTREAM SKIPPED WRITE: %s", resourceName) + } err = srv.SendAndClose(&resp) if err != nil { diff --git a/server/grpc_test.go b/server/grpc_test.go index 2e9eda2..74aec6c 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -3106,7 +3106,7 @@ func TestInsufficientStorageWhenProxyTriesToStoreAc(t *testing.T) { } } -func TestGrpcByteStreamWriteShedWhenInflightCapReached(t *testing.T) { +func TestGrpcByteStreamWriteAbsorbedWhenInflightCapReached(t *testing.T) { fixture := grpcTestSetupInternal(t, false, WithMaxInflightCASWrites(1)) defer os.RemoveAll(fixture.tempdir) @@ -3146,14 +3146,14 @@ func TestGrpcByteStreamWriteShedWhenInflightCapReached(t *testing.T) { // Let the server consume the first chunk and take the only slot. time.Sleep(200 * time.Millisecond) + // With the only slot held, a new blob is absorbed: the client sees + // success, but nothing is stored. otherBlob, otherHash := testutils.RandomDataAndHash(64) - err = fullWrite(otherBlob, otherHash) - if status.Code(err) != codes.Unavailable { - t.Fatalf("expected codes.Unavailable while a write holds the slot, got %v", err) + if err := fullWrite(otherBlob, otherHash); err != nil { + t.Fatalf("absorbed write should be acknowledged, got %v", err) } - - if err := fullWrite(otherBlob, otherHash); status.Code(err) != codes.Unavailable { - t.Fatalf("expected second shed, got %v", err) + if ok, _ := fixture.diskCache.Contains(ctx, cache.CAS, otherHash, int64(len(otherBlob))); ok { + t.Fatal("absorbed blob must not be stored") } err = held.Send(&bytestream.WriteRequest{ @@ -3172,6 +3172,9 @@ func TestGrpcByteStreamWriteShedWhenInflightCapReached(t *testing.T) { if err := fullWrite(otherBlob, otherHash); err != nil { t.Fatalf("write after slot release should succeed: %v", err) } + if ok, _ := fixture.diskCache.Contains(ctx, cache.CAS, otherHash, int64(len(otherBlob))); !ok { + t.Fatal("write after slot release must be stored") + } // Now present: the Contains early return answers before the gate. if err := fullWrite(otherBlob, otherHash); err != nil { t.Fatalf("duplicate write of present blob should succeed: %v", err) From fb5e3f322ebb115eb9644dc11aa2e7fae71be043 Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 00:59:37 +0000 Subject: [PATCH 7/9] Revert "server: acknowledge new CAS writes at the admission gate instead of returning UNAVAILABLE" This reverts commit 53862ab0281f6161d930cb8f15bbd1eca0fb0c3e. --- server/grpc_bytestream.go | 27 +++++++++------------------ server/grpc_test.go | 17 +++++++---------- 2 files changed, 16 insertions(+), 28 deletions(-) diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go index 81d1e58..f24e1fd 100644 --- a/server/grpc_bytestream.go +++ b/server/grpc_bytestream.go @@ -431,10 +431,7 @@ var errWriteOffset error = errors.New("bytestream writes from non-zero offsets a // UNAVAILABLE is the connection-class code clients treat as "drop this // write", never a failed build. -// errCASWriteAbsorbed signals a new CAS write that was acknowledged at the -// admission gate without being stored: the client sees committed_size as if -// the blob existed and stops sending, and the blob is simply a future miss. -var errCASWriteAbsorbed = errors.New("cas write absorbed at admission gate") +var errCASWriteShed = status.Error(codes.Unavailable, "too many in-flight CAS writes") var errDecoderPoolFail error = errors.New("failed to get DecoderWrapper from pool") func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { @@ -561,12 +558,8 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { // release. if !s.tryAcquireCASWriteSlot() { _ = rc.Close() - if cmp == casblob.Identity { - resp.CommittedSize = size - } else { - resp.CommittedSize = -1 - } - putResult <- errCASWriteAbsorbed + s.accessLogger.Printf("GRPC BYTESTREAM WRITE SHED: %s", resourceName) + recvResult <- errCASWriteShed return } @@ -647,8 +640,10 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { } _ = pw.CloseWithError(err) - s.accessLogger.Printf("GRPC BYTESTREAM WRITE FAILED: %s %s", - resourceName, err.Error()) + if err != errCASWriteShed { + s.accessLogger.Printf("GRPC BYTESTREAM WRITE FAILED: %s %s", + resourceName, err.Error()) + } return err } @@ -658,12 +653,8 @@ func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { default: } - if err == io.EOF || err == errCASWriteAbsorbed { - if err == errCASWriteAbsorbed { - s.accessLogger.Printf("GRPC BYTESTREAM WRITE ABSORBED: %s", resourceName) - } else { - s.accessLogger.Printf("GRPC BYTESTREAM SKIPPED WRITE: %s", resourceName) - } + if err == io.EOF { + s.accessLogger.Printf("GRPC BYTESTREAM SKIPPED WRITE: %s", resourceName) err = srv.SendAndClose(&resp) if err != nil { diff --git a/server/grpc_test.go b/server/grpc_test.go index 74aec6c..2e9eda2 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -3106,7 +3106,7 @@ func TestInsufficientStorageWhenProxyTriesToStoreAc(t *testing.T) { } } -func TestGrpcByteStreamWriteAbsorbedWhenInflightCapReached(t *testing.T) { +func TestGrpcByteStreamWriteShedWhenInflightCapReached(t *testing.T) { fixture := grpcTestSetupInternal(t, false, WithMaxInflightCASWrites(1)) defer os.RemoveAll(fixture.tempdir) @@ -3146,14 +3146,14 @@ func TestGrpcByteStreamWriteAbsorbedWhenInflightCapReached(t *testing.T) { // Let the server consume the first chunk and take the only slot. time.Sleep(200 * time.Millisecond) - // With the only slot held, a new blob is absorbed: the client sees - // success, but nothing is stored. otherBlob, otherHash := testutils.RandomDataAndHash(64) - if err := fullWrite(otherBlob, otherHash); err != nil { - t.Fatalf("absorbed write should be acknowledged, got %v", err) + err = fullWrite(otherBlob, otherHash) + if status.Code(err) != codes.Unavailable { + t.Fatalf("expected codes.Unavailable while a write holds the slot, got %v", err) } - if ok, _ := fixture.diskCache.Contains(ctx, cache.CAS, otherHash, int64(len(otherBlob))); ok { - t.Fatal("absorbed blob must not be stored") + + if err := fullWrite(otherBlob, otherHash); status.Code(err) != codes.Unavailable { + t.Fatalf("expected second shed, got %v", err) } err = held.Send(&bytestream.WriteRequest{ @@ -3172,9 +3172,6 @@ func TestGrpcByteStreamWriteAbsorbedWhenInflightCapReached(t *testing.T) { if err := fullWrite(otherBlob, otherHash); err != nil { t.Fatalf("write after slot release should succeed: %v", err) } - if ok, _ := fixture.diskCache.Contains(ctx, cache.CAS, otherHash, int64(len(otherBlob))); !ok { - t.Fatal("write after slot release must be stored") - } // Now present: the Contains early return answers before the gate. if err := fullWrite(otherBlob, otherHash); err != nil { t.Fatalf("duplicate write of present blob should succeed: %v", err) From cd28524411de6827ca78bb656b98dc5cdf785d2c Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 01:42:24 +0000 Subject: [PATCH 8/9] server: shed gated CAS writes with FAILED_PRECONDITION and export gate metrics Bazel's RemoteRetrier classifies UNAVAILABLE as a transient failure, so a shed Write cost the client the full --remote_retries backoff ladder per blob and showed up as a multi-minute upload tail after the build summary. FAILED_PRECONDITION is a permanent failure for the client: the upload is reported once and the invocation moves on. The action result is still not published, so nothing references an unstored blob. bazel_remote_cas_write_slots_inflight and bazel_remote_cas_write_shed_total expose the gate directly; grpc started minus handled on Write also counts existing-blob early returns and sheds in flight. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- server/grpc_bytestream.go | 2 +- server/grpc_test.go | 6 +++--- server/runtime.go | 15 +++++++++++++++ 3 files changed, 19 insertions(+), 4 deletions(-) diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go index f24e1fd..a1b5b90 100644 --- a/server/grpc_bytestream.go +++ b/server/grpc_bytestream.go @@ -431,7 +431,7 @@ var errWriteOffset error = errors.New("bytestream writes from non-zero offsets a // UNAVAILABLE is the connection-class code clients treat as "drop this // write", never a failed build. -var errCASWriteShed = status.Error(codes.Unavailable, "too many in-flight CAS writes") +var errCASWriteShed = status.Error(codes.FailedPrecondition, "too many in-flight CAS writes") var errDecoderPoolFail error = errors.New("failed to get DecoderWrapper from pool") func (s *grpcServer) Write(srv bytestream.ByteStream_WriteServer) error { diff --git a/server/grpc_test.go b/server/grpc_test.go index 2e9eda2..68385cd 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -3148,11 +3148,11 @@ func TestGrpcByteStreamWriteShedWhenInflightCapReached(t *testing.T) { otherBlob, otherHash := testutils.RandomDataAndHash(64) err = fullWrite(otherBlob, otherHash) - if status.Code(err) != codes.Unavailable { - t.Fatalf("expected codes.Unavailable while a write holds the slot, got %v", err) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("expected codes.FailedPrecondition while a write holds the slot, got %v", err) } - if err := fullWrite(otherBlob, otherHash); status.Code(err) != codes.Unavailable { + if err := fullWrite(otherBlob, otherHash); status.Code(err) != codes.FailedPrecondition { t.Fatalf("expected second shed, got %v", err) } diff --git a/server/runtime.go b/server/runtime.go index c0465e4..f863664 100644 --- a/server/runtime.go +++ b/server/runtime.go @@ -3,6 +3,8 @@ package server import ( "context" "fmt" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "sync" "time" @@ -95,14 +97,26 @@ func WithMaxInflightCASWrites(n int) GRPCServerOption { } } +var casWriteSlotsInflight = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "bazel_remote_cas_write_slots_inflight", + Help: "New CAS ByteStream writes currently admitted through the in-flight write gate.", +}) + +var casWriteShedTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bazel_remote_cas_write_shed_total", + Help: "New CAS ByteStream writes rejected because the in-flight write gate was full.", +}) + func (s *grpcServer) tryAcquireCASWriteSlot() bool { if s.casWriteSlots == nil { return true } select { case s.casWriteSlots <- struct{}{}: + casWriteSlotsInflight.Inc() return true default: + casWriteShedTotal.Inc() return false } } @@ -110,6 +124,7 @@ func (s *grpcServer) tryAcquireCASWriteSlot() bool { func (s *grpcServer) releaseCASWriteSlot() { if s.casWriteSlots != nil { <-s.casWriteSlots + casWriteSlotsInflight.Dec() } } From 5325705bc4c9071a73366982bf29979a3fff3aed Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Fri, 9 Oct 2026 01:49:26 +0000 Subject: [PATCH 9/9] server: document the FAILED_PRECONDITION shed contract Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- server/grpc_bytestream.go | 8 ++++++-- server/runtime.go | 2 +- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go index a1b5b90..04e42a4 100644 --- a/server/grpc_bytestream.go +++ b/server/grpc_bytestream.go @@ -429,8 +429,12 @@ func (s *grpcServer) parseWriteResource(r string) (string, int64, casblob.Compre var errWriteOffset error = errors.New("bytestream writes from non-zero offsets are unsupported") -// UNAVAILABLE is the connection-class code clients treat as "drop this -// write", never a failed build. +// FAILED_PRECONDITION is deliberately not a connection-class code: Bazel's +// retrier treats UNAVAILABLE as transient and walks the full --remote_retries +// backoff ladder per blob, which under sustained shedding becomes a +// minutes-long upload tail after the build summary. A permanent code makes +// the client log the upload failure once and move on; the action result is +// never published, so nothing references the unstored blob. var errCASWriteShed = status.Error(codes.FailedPrecondition, "too many in-flight CAS writes") var errDecoderPoolFail error = errors.New("failed to get DecoderWrapper from pool") diff --git a/server/runtime.go b/server/runtime.go index f863664..607fc5e 100644 --- a/server/runtime.go +++ b/server/runtime.go @@ -84,7 +84,7 @@ func WithReadChunkSizeBytes(chunkSizeBytes int64) GRPCServerOption { } // WithMaxInflightCASWrites caps concurrent CAS ByteStream Writes that reach -// the Put path; excess writes are shed with UNAVAILABLE. 0 = unlimited. +// the Put path; excess writes are shed with FAILED_PRECONDITION. 0 = unlimited. func WithMaxInflightCASWrites(n int) GRPCServerOption { return func(s *grpcServer) error { if n < 0 {