internal/pkg/rsproxy/proxy.go - rsclient - Git at Google

 // Copyright 2026 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 package rsproxy

 import (
 	"context"
 	"fmt"
 	"net"
 	"sync"
 	"time"

 	"github.com/bazelbuild/rsclient/internal/pkg/auth"
 	"github.com/bazelbuild/rsclient/internal/pkg/fifo"
 	"github.com/bazelbuild/rsclient/internal/pkg/syncutil"
 	"github.com/golang/glog"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 )

 // RPCTimeouts is a map that defines custom timeouts for specific RPC operations.
 type RPCTimeouts map[string]time.Duration

 // DefaultRPCTimeouts provides conservative default timeouts for backend RPC calls.
 var DefaultRPCTimeouts = map[string]time.Duration{
 	"default": 20 * time.Second,
 }

 // ProxyEgressParams holds the gRPC connections required to initialize a Proxy's egress workers.
 type ProxyEgressParams struct {
 	RSConn  *grpc.ClientConn
 	CASConn *grpc.ClientConn
 	JSConn  *grpc.ClientConn
 }

 // ProxyConfig defines the complete configuration for a Proxy and its workers.
 type ProxyConfig struct {
 	ResultStoreEgressOptions
 	JobStatusEgressOptions
 	PrototextWriterOptions

 	StreamIngressOptions
 	GRPCIngressOptions
 	PrototextReaderOptions

 	// IngressListener is an optional existing listener for the gRPC ingress.
 	// Intended for testing or specialized embedding as a dependency injection point.
 	IngressListener net.Listener

 	// RSConn, CASConn, and JSConn are optional existing connections (e.g. for testing).
 	// If set, these take precedence over dialing new connections from RSDialParams, etc.
 	RSConn  *grpc.ClientConn
 	CASConn *grpc.ClientConn
 	JSConn  *grpc.ClientConn

 	// RPCTimeouts sets deadlines for individual RPC calls.
 	RPCTimeouts RPCTimeouts

 	// Retrier is the policy used for retrying transient RPC failures.
 	Retrier *Retrier
 }

 // HasGRPCIngress returns true if a gRPC ingress worker is configured.
 func (c *ProxyConfig) HasGRPCIngress() bool {
 	return c.GRPCIngressOptions.Addr != "" || c.IngressListener != nil
 }

 // Proxy is a high-level orchestrator for communicating with multiple backend
 // services (e.g. ResultStore, CAS, JobStatus) and managing the build event ingress.
 //
 // It manages the lifecycle of specialized egress workers and uses a Router to
 // fan out incoming build events from various ingress sources (e.g. FIFO, gRPC).
 //
 // The Proxy acts as the central coordinator for the entire build event pipeline,
 // ensuring that ingress sources are properly opened, events are correctly
 // routed to all active egress workers, and that the backend invocation is
 // gracefully finalized after the build tool finishes.
 type Proxy struct {
 	// config holds the validated configuration used to initialize the proxy.
 	config *ProxyConfig

 	// creds are common credential options used for all underlying RPC calls.
 	creds credentials.PerRPCCredentials

 	// router fans out build events to registered egress workers.
 	router *Router

 	// ingressWorker is the configured input source (e.g., FIFO, gRPC, or Prototext).
 	ingressWorker IngressWorker

 	// FinalizationSignal is closed when the client should begin processing
 	// post-build artifacts and finalize the backend invocation.
 	FinalizationSignal chan struct{}

 	// runStarted is closed when the Run() function has successfully started
 	// its background processing pipeline.
 	runStarted     chan struct{}
 	runStartedOnce sync.Once

 	// ready provides a thread-safe way to broadcast the proxy's runtime
 	// metadata to multiple concurrent listeners.
 	ready syncutil.ReadySignal[IngressMetadata]

 	// cleanupFns is a list of functions to be called during shutdown to
 	// release resources (e.g., closing file handles, stopping background tasks).
 	cleanupFns []func()

 	// cleanupDone is closed when all cleanup functions have finished executing.
 	cleanupDone chan bool

 	// rrOnce ensures that DrainAndReleaseResources is only executed once.
 	rrOnce sync.Once
 	// fOnce ensures that SignalFinalization is only executed once.
 	fOnce sync.Once
 }

 // IngressMetadata contains information about the proxy's active ingress.
 type IngressMetadata struct {
 	// GRPCAddr is the network address the gRPC ingress is bound to.
 	GRPCAddr string
 	// FIFOPath is the filesystem path to the FIFO/File ingress.
 	FIFOPath string
 }

 // SignalReadyFunc is a callback triggered when a component is ready to
 // process traffic.
 type SignalReadyFunc func(IngressMetadata)

 // NewProxyFromConnection creates a Proxy from existing gRPC connections.
 // It initializes and registers the ResultStore and JobStatus workers with the Router.
 func NewProxyFromConnection(ctx context.Context, p ProxyEgressParams, opts ...ProxyOpt) (*Proxy, error) {
 	connOpt := &connectionOpt{params: p}
 	allOpts := make([]ProxyOpt, 0, len(opts)+1)
 	allOpts = append(allOpts, connOpt)
 	allOpts = append(allOpts, opts...)
 	return NewProxy(ctx, allOpts...)
 }

 // connectionOpt is an internal ProxyOpt used to inject existing connections into the ProxyConfig.
 type connectionOpt struct {
 	params ProxyEgressParams
 }

 func (o *connectionOpt) ApplyProxy(p *Proxy) {
 	p.config.RSConn = o.params.RSConn
 	p.config.CASConn = o.params.CASConn
 	p.config.JSConn = o.params.JSConn
 }

 // ProxyOpt is an interface for configuring the Proxy during initialization.
 type ProxyOpt interface {
 	ApplyProxy(*Proxy)
 }

 // ApplyProxy implements ProxyOpt for RPCTimeouts.
 func (d RPCTimeouts) ApplyProxy(p *Proxy) {
 	p.config.RPCTimeouts = d
 }

 // ApplyProxy implements ProxyOpt for Retrier.
 func (r *Retrier) ApplyProxy(p *Proxy) {
 	p.config.Retrier = r
 }

 // ApplyProxy implements ProxyOpt for ProxyConfig.
 func (c *ProxyConfig) ApplyProxy(p *Proxy) {
 	// Surgically merge the provided options into the proxy's internal config.
 	if c.RSEgress != nil {
 		p.config.RSEgress = c.RSEgress
 	}
 	if c.CASEgress != nil {
 		p.config.CASEgress = c.CASEgress
 	}
 	if c.Conn != nil {
 		p.config.Conn = c.Conn
 	}

 	if c.RSInstanceName != "" {
 		p.config.RSInstanceName = c.RSInstanceName
 	}
 	if c.CASInstanceName != "" {
 		p.config.CASInstanceName = c.CASInstanceName
 	}
 	if c.PartialInvocation != nil {
 		p.config.PartialInvocation = c.PartialInvocation
 	}
 	if c.ResultsURL != "" {
 		p.config.ResultsURL = c.ResultsURL
 	}
 	if c.QuietMode {
 		p.config.QuietMode = true
 	}
 	if len(c.PreBuildArtifacts) > 0 {
 		p.config.PreBuildArtifacts = c.PreBuildArtifacts
 	}
 	if len(c.PostBuildArtifacts) > 0 {
 		p.config.PostBuildArtifacts = c.PostBuildArtifacts
 	}
 	if c.BundleCountThreshold != 0 {
 		p.config.BundleCountThreshold = c.BundleCountThreshold
 	}
 	if c.BundleByteThreshold != 0 {
 		p.config.BundleByteThreshold = c.BundleByteThreshold
 	}
 	if c.MangleResourceIDs {
 		p.config.MangleResourceIDs = true
 	}
 	if c.Passthrough {
 		p.config.Passthrough = true
 	}

 	if c.JobStatusAddr != "" {
 		p.config.JobStatusAddr = c.JobStatusAddr
 	}

 	if c.PrototextWriterOptions.Path != "" {
 		p.config.PrototextWriterOptions.Path = c.PrototextWriterOptions.Path
 	}

 	if c.StreamIngressOptions.InputFile != "" {
 		p.config.SetFile(c.StreamIngressOptions.InputFile)
 	} else if c.StreamIngressOptions.InputFIFO != "" {
 		p.config.SetFIFO(c.StreamIngressOptions.InputFIFO)
 	}
 	if c.StreamIngressOptions.IsTemp {
 		p.config.StreamIngressOptions.IsTemp = true
 	}
 	if c.StreamIngressOptions.OpenBlocking {
 		p.config.StreamIngressOptions.OpenBlocking = true
 	}
 	if c.GRPCIngressOptions.Addr != "" {
 		p.config.GRPCIngressOptions.Addr = c.GRPCIngressOptions.Addr
 	}
 	if c.PrototextReaderOptions.Path != "" {
 		p.config.PrototextReaderOptions.Path = c.PrototextReaderOptions.Path
 	}

 	if c.IngressListener != nil {
 		p.config.IngressListener = c.IngressListener
 	}

 	if c.RSConn != nil {
 		p.config.RSConn = c.RSConn
 	}
 	if c.CASConn != nil {
 		p.config.CASConn = c.CASConn
 	}
 	if c.JSConn != nil {
 		p.config.JSConn = c.JSConn
 	}
 	if c.RPCTimeouts != nil {
 		p.config.RPCTimeouts = c.RPCTimeouts
 	}
 	if c.Retrier != nil {
 		p.config.Retrier = c.Retrier
 	}
 }

 // ProxyPerRPCCredsOpt is an ProxyOpt that provides per-RPC credentials to the Proxy.
 type ProxyPerRPCCredsOpt struct {
 	Creds credentials.PerRPCCredentials
 }

 // ApplyProxy implements ProxyOpt for ProxyPerRPCCredsOpt.
 func (p *ProxyPerRPCCredsOpt) ApplyProxy(proxy *Proxy) {
 	proxy.creds = p.Creds
 }

 // DialParamsOpt is an ProxyOpt that provides dial parameters for ResultStore, CAS, and JobStatus.
 type DialParamsOpt struct {
 	RSDialParams  auth.DialParams
 	CASDialParams auth.DialParams
 	JSDialParams  auth.DialParams
 }

 // ApplyProxy implements ProxyOpt for DialParamsOpt.
 func (d *DialParamsOpt) ApplyProxy(p *Proxy) {
 	if d.RSDialParams.Service != "" {
 		p.config.RSEgress = auth.NewServiceConnection(d.RSDialParams)
 	}
 	if d.CASDialParams.Service != "" {
 		p.config.CASEgress = auth.NewServiceConnection(d.CASDialParams)
 	}
 	if d.JSDialParams.Service != "" {
 		p.config.Conn = auth.NewServiceConnection(d.JSDialParams)
 	}
 }

 // NewProxy creates a new Proxy and establishes connections to the ResultStore,
 // CAS, and (optionally) JobStatus services based on the provided configuration.
 func NewProxy(ctx context.Context, opts ...ProxyOpt) (*Proxy, error) {
 	proxy := &Proxy{
 		config: &ProxyConfig{
 			ResultStoreEgressOptions: ResultStoreEgressOptions{
 				BundleCountThreshold: DefaultBundleCountThreshold,
 				BundleByteThreshold:  DefaultBundleByteThreshold,
 			},
 			RPCTimeouts: DefaultRPCTimeouts,
 			Retrier:     RetryTransient(),
 		},
 		router:             NewRouter(ctx),
 		FinalizationSignal: make(chan struct{}),
 		runStarted:         make(chan struct{}),
 		cleanupDone:        make(chan bool),
 	}

 	// Phase 1: Apply all options to the configuration.
 	for _, o := range opts {
 		if o != nil {
 			o.ApplyProxy(proxy)
 		}
 	}

 	// Phase 2: Initialize workers based on the final configuration.
 	// ResultStore Worker
 	if w, err := proxy.config.ResultStoreEgressOptions.CreateWorker(ctx, proxy.config.RSConn, proxy.config.CASConn, proxy.config.Retrier, proxy.RPCOpts, proxy.FinalizationSignal); err != nil {
 		return nil, err
 	} else if w != nil {
 		if err := proxy.router.RegisterEgressWorker(w); err != nil {
 			proxy.Close(ctx)
 			return nil, err
 		}
 	}

 	// JobStatus Worker
 	if w, err := proxy.config.JobStatusEgressOptions.CreateWorker(ctx, proxy.config.JSConn, proxy.config.Retrier, proxy.RPCOpts); err != nil {
 		return nil, err
 	} else if w != nil {
 		if err := proxy.router.RegisterEgressWorker(w); err != nil {
 			proxy.Close(ctx)
 			return nil, err
 		}
 	}

 	// PrototextWriter
 	if w, err := proxy.config.PrototextWriterOptions.CreateWorker(); err != nil {
 		return nil, err
 	} else if w != nil {
 		if err := proxy.router.RegisterEgressWorker(w); err != nil {
 			proxy.Close(ctx)
 			return nil, err
 		}
 	}

 	// Phase 3: Initialize Ingress Worker
 	if proxy.config.HasGRPCIngress() {
 		lis := proxy.config.IngressListener
 		if lis == nil && proxy.config.GRPCIngressOptions.Addr != "bufnet" {
 			var err error
 			lis, err = net.Listen("tcp", proxy.config.GRPCIngressOptions.Addr)
 			if err != nil {
 				return nil, err
 			}
 			proxy.config.GRPCIngressOptions.Addr = lis.Addr().String()
 		}
 		proxy.ingressWorker = NewGRPCIngressWorker(lis, proxy.signalIngressReady)
 	} else if proxy.config.PrototextReaderOptions.Path != "" {
 		proxy.ingressWorker = NewPrototextReader(proxy.config.PrototextReaderOptions.Path)
 	} else if proxy.config.StreamIngressOptions.InputFIFO != "" || proxy.config.StreamIngressOptions.InputFile != "" || proxy.config.StreamIngressOptions.IsTemp {
 		worker := NewStreamIngressWorker(proxy.config.StreamIngressOptions.ToInputSource(), proxy.signalIngressReady)

 		// Setup the source (creates it on disk and potentially blocks on open).
 		cleanup, err := worker.(*streamIngressWorker).Setup(ctx)
 		if err != nil {
 			proxy.Close(ctx)
 			return nil, fmt.Errorf("failed to setup stream ingress: %w", err)
 		}

 		// Update the path in config (handles random temp names and input-type specific fields).
 		path := worker.(*streamIngressWorker).Path()
 		if proxy.config.StreamIngressOptions.InputFile != "" {
 			proxy.config.StreamIngressOptions.InputFile = path
 		} else {
 			proxy.config.StreamIngressOptions.InputFIFO = path
 		}
 		proxy.ingressWorker = worker

 		// Signal readiness immediately since the FIFO/File is now created on disk.
 		proxy.signalIngressReady(IngressMetadata{FIFOPath: path})

 		// Ensure the source is cleaned up when the proxy is closed.
 		proxy.AddCleanup(cleanup)
 	}

 	return proxy, nil
 }

 // Close gracefully shuts down the proxy and all its registered workers.
 func (p *Proxy) Close(ctx context.Context) error {
 	glog.Infof("proxy waiting for background tasks to finish")
 	select {
 	case <-p.runStarted:
 		// Run was called, wait for the router to finish.
 	default:
 		// Run was never called, do nothing.
 		return nil
 	}
 	glog.Infof("proxy background tasks finished")

 	return p.router.Close(ctx)
 }

 // Run executes the internal Router, which kicks off all registered workers
 // in parallel. It returns when the ingress stream is finished and all workers
 // have successfully completed their processing.
 func (p *Proxy) Run(ctx context.Context) error {
 	p.runStartedOnce.Do(func() {
 		close(p.runStarted)
 	})
 	worker := p.ingressWorker
 	if worker == nil {
 		worker = noopIngressWorker{}
 		p.signalIngressReady(IngressMetadata{})
 	}
 	if err := p.router.SetIngress(worker); err != nil {
 		return err
 	}
 	return p.router.Run(ctx)
 }

 // Ready returns a channel that will receive IngressMetadata once the server
 // is configured and ready to process build events. This method is
 // thread-safe and supports multiple concurrent listeners (broadcast) as
 // well as late-joiners.
 func (p *Proxy) Ready() <-chan IngressMetadata {
 	return p.ready.Ready()
 }

 // IngressDrainedChan returns a channel that is closed when the ingress
 // source has been completely consumed and drained by the dispatcher.
 func (p *Proxy) IngressDrainedChan() <-chan struct{} {
 	return p.router.IngressDrainedChan()
 }

 // signalIngressReady signals that the proxy is ready to accept requests.
 func (p *Proxy) signalIngressReady(m IngressMetadata) {
 	p.ready.Signal(m)
 }

 // SignalIngressEOF unblocks the proxy's ingress reader by sending a best-effort
 // EOF signal. This is primarily used for FIFO ingress to ensure the reader
 // loop doesn't hang if the build tool exits without ever opening the pipe.
 // It is a no-op if the ingress source is not a FIFO.
 func (p *Proxy) SignalIngressEOF() error {
 	var m IngressMetadata
 	if v, ok := p.ready.Value(); ok {
 		m = v
 	}
 	if m.FIFOPath != "" {
 		return fifo.SignalFIFOEOF(m.FIFOPath)
 	}
 	return nil
 }

 // SignalFinalization signals to all workers that the primary build phase is
 // complete and they should process post-build artifacts and finalize their
 // respective backend operations.
 func (p *Proxy) SignalFinalization() {
 	p.fOnce.Do(func() {
 		glog.Infof("Signaling workers to finalize invocation and upload post-build artifacts.")
 		close(p.FinalizationSignal)
 	})
 }

 // AddCleanup adds a function to be called during DrainAndReleaseResources.
 func (p *Proxy) AddCleanup(fn func()) {
 	p.cleanupFns = append(p.cleanupFns, fn)
 }

 // DrainAndReleaseResources runs all registered cleanup functions.
 func (p *Proxy) DrainAndReleaseResources() {
 	p.rrOnce.Do(func() {
 		glog.Infof("Proxy shutting down.")
 		defer p.ready.Close()
 		defer func() {
 			glog.Infof("Proxy shutdown complete.")
 			close(p.cleanupDone)
 		}()
 		for _, cleanup := range p.cleanupFns {
 			if cleanup != nil {
 				cleanup()
 			}
 		}
 	})
 }

 // Shutdown initiates a graceful shutdown of the proxy, cleaning up all resources.
 func (p *Proxy) Shutdown() {
 	p.DrainAndReleaseResources()
 	<-p.cleanupDone
 }

 // Router returns the internal Router instance managed by this proxy.
 func (p *Proxy) Router() *Router {
 	return p.router
 }

 // Errors returns a snapshot of the non-fatal errors encountered
 // across all active workers.
 func (p *Proxy) Errors() []error {
 	return p.router.Errors()
 }

 // HasEgress returns true if at least one egress worker is configured and ready to run.
 func (p *Proxy) HasEgress() bool {
 	return p.router.HasEgress()
 }

 // HasGRPCIngress returns true if a gRPC ingress worker is configured.
 func (p *Proxy) HasGRPCIngress() bool {
 	return p.config.HasGRPCIngress()
 }

 // RPCOpts returns the set of default gRPC call options (like authentication)
 // that should be applied to all requests made by this proxy.
 func (p *Proxy) RPCOpts() []grpc.CallOption {
 	opts := []grpc.CallOption{}
 	if p.creds == nil {
 		return opts
 	}
 	return append(opts, grpc.PerRPCCredentials(p.creds))
 }

 // CallWithTimeout wraps a gRPC call with the configured timeout for that
 // specific operation.
 func (p *Proxy) CallWithTimeout(ctx context.Context, rpcName string, f func(ctx context.Context) error) error {
 	timeout, ok := p.config.RPCTimeouts[rpcName]
 	if !ok {
 		if timeout, ok = p.config.RPCTimeouts["default"]; !ok {
 			timeout = 0
 		}
 	}
 	if timeout == 0 {
 		return f(ctx)
 	}
 	childCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 	return f(childCtx)
 }
	// Copyright 2026 Google LLC
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	package rsproxy

	import (
	"context"
	"fmt"
	"net"
	"sync"
	"time"

	"github.com/bazelbuild/rsclient/internal/pkg/auth"
	"github.com/bazelbuild/rsclient/internal/pkg/fifo"
	"github.com/bazelbuild/rsclient/internal/pkg/syncutil"
	"github.com/golang/glog"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	)

	// RPCTimeouts is a map that defines custom timeouts for specific RPC operations.
	type RPCTimeouts map[string]time.Duration

	// DefaultRPCTimeouts provides conservative default timeouts for backend RPC calls.
	var DefaultRPCTimeouts = map[string]time.Duration{
	"default": 20 * time.Second,
	}

	// ProxyEgressParams holds the gRPC connections required to initialize a Proxy's egress workers.
	type ProxyEgressParams struct {
	RSConn *grpc.ClientConn
	CASConn *grpc.ClientConn
	JSConn *grpc.ClientConn
	}

	// ProxyConfig defines the complete configuration for a Proxy and its workers.
	type ProxyConfig struct {
	ResultStoreEgressOptions
	JobStatusEgressOptions
	PrototextWriterOptions

	StreamIngressOptions
	GRPCIngressOptions
	PrototextReaderOptions

	// IngressListener is an optional existing listener for the gRPC ingress.
	// Intended for testing or specialized embedding as a dependency injection point.
	IngressListener net.Listener

	// RSConn, CASConn, and JSConn are optional existing connections (e.g. for testing).
	// If set, these take precedence over dialing new connections from RSDialParams, etc.
	RSConn *grpc.ClientConn
	CASConn *grpc.ClientConn
	JSConn *grpc.ClientConn

	// RPCTimeouts sets deadlines for individual RPC calls.
	RPCTimeouts RPCTimeouts

	// Retrier is the policy used for retrying transient RPC failures.
	Retrier *Retrier
	}

	// HasGRPCIngress returns true if a gRPC ingress worker is configured.
	func (c *ProxyConfig) HasGRPCIngress() bool {
	return c.GRPCIngressOptions.Addr != "" \|\| c.IngressListener != nil
	}

	// Proxy is a high-level orchestrator for communicating with multiple backend
	// services (e.g. ResultStore, CAS, JobStatus) and managing the build event ingress.
	//
	// It manages the lifecycle of specialized egress workers and uses a Router to
	// fan out incoming build events from various ingress sources (e.g. FIFO, gRPC).
	//
	// The Proxy acts as the central coordinator for the entire build event pipeline,
	// ensuring that ingress sources are properly opened, events are correctly
	// routed to all active egress workers, and that the backend invocation is
	// gracefully finalized after the build tool finishes.
	type Proxy struct {
	// config holds the validated configuration used to initialize the proxy.
	config *ProxyConfig

	// creds are common credential options used for all underlying RPC calls.
	creds credentials.PerRPCCredentials

	// router fans out build events to registered egress workers.
	router *Router

	// ingressWorker is the configured input source (e.g., FIFO, gRPC, or Prototext).
	ingressWorker IngressWorker

	// FinalizationSignal is closed when the client should begin processing
	// post-build artifacts and finalize the backend invocation.
	FinalizationSignal chan struct{}

	// runStarted is closed when the Run() function has successfully started
	// its background processing pipeline.
	runStarted chan struct{}
	runStartedOnce sync.Once

	// ready provides a thread-safe way to broadcast the proxy's runtime
	// metadata to multiple concurrent listeners.
	ready syncutil.ReadySignal[IngressMetadata]

	// cleanupFns is a list of functions to be called during shutdown to
	// release resources (e.g., closing file handles, stopping background tasks).
	cleanupFns []func()

	// cleanupDone is closed when all cleanup functions have finished executing.
	cleanupDone chan bool

	// rrOnce ensures that DrainAndReleaseResources is only executed once.
	rrOnce sync.Once
	// fOnce ensures that SignalFinalization is only executed once.
	fOnce sync.Once
	}

	// IngressMetadata contains information about the proxy's active ingress.
	type IngressMetadata struct {
	// GRPCAddr is the network address the gRPC ingress is bound to.
	GRPCAddr string
	// FIFOPath is the filesystem path to the FIFO/File ingress.
	FIFOPath string
	}

	// SignalReadyFunc is a callback triggered when a component is ready to
	// process traffic.
	type SignalReadyFunc func(IngressMetadata)

	// NewProxyFromConnection creates a Proxy from existing gRPC connections.
	// It initializes and registers the ResultStore and JobStatus workers with the Router.
	func NewProxyFromConnection(ctx context.Context, p ProxyEgressParams, opts ...ProxyOpt) (*Proxy, error) {
	connOpt := &connectionOpt{params: p}
	allOpts := make([]ProxyOpt, 0, len(opts)+1)
	allOpts = append(allOpts, connOpt)
	allOpts = append(allOpts, opts...)
	return NewProxy(ctx, allOpts...)
	}

	// connectionOpt is an internal ProxyOpt used to inject existing connections into the ProxyConfig.
	type connectionOpt struct {
	params ProxyEgressParams
	}

	func (o connectionOpt) ApplyProxy(p Proxy) {
	p.config.RSConn = o.params.RSConn
	p.config.CASConn = o.params.CASConn
	p.config.JSConn = o.params.JSConn
	}

	// ProxyOpt is an interface for configuring the Proxy during initialization.
	type ProxyOpt interface {
	ApplyProxy(*Proxy)
	}

	// ApplyProxy implements ProxyOpt for RPCTimeouts.
	func (d RPCTimeouts) ApplyProxy(p *Proxy) {
	p.config.RPCTimeouts = d
	}

	// ApplyProxy implements ProxyOpt for Retrier.
	func (r Retrier) ApplyProxy(p Proxy) {
	p.config.Retrier = r
	}

	// ApplyProxy implements ProxyOpt for ProxyConfig.
	func (c ProxyConfig) ApplyProxy(p Proxy) {
	// Surgically merge the provided options into the proxy's internal config.
	if c.RSEgress != nil {
	p.config.RSEgress = c.RSEgress
	}
	if c.CASEgress != nil {
	p.config.CASEgress = c.CASEgress
	}
	if c.Conn != nil {
	p.config.Conn = c.Conn
	}

	if c.RSInstanceName != "" {
	p.config.RSInstanceName = c.RSInstanceName
	}
	if c.CASInstanceName != "" {
	p.config.CASInstanceName = c.CASInstanceName
	}
	if c.PartialInvocation != nil {
	p.config.PartialInvocation = c.PartialInvocation
	}
	if c.ResultsURL != "" {
	p.config.ResultsURL = c.ResultsURL
	}
	if c.QuietMode {
	p.config.QuietMode = true
	}
	if len(c.PreBuildArtifacts) > 0 {
	p.config.PreBuildArtifacts = c.PreBuildArtifacts
	}
	if len(c.PostBuildArtifacts) > 0 {
	p.config.PostBuildArtifacts = c.PostBuildArtifacts
	}
	if c.BundleCountThreshold != 0 {
	p.config.BundleCountThreshold = c.BundleCountThreshold
	}
	if c.BundleByteThreshold != 0 {
	p.config.BundleByteThreshold = c.BundleByteThreshold
	}
	if c.MangleResourceIDs {
	p.config.MangleResourceIDs = true
	}
	if c.Passthrough {
	p.config.Passthrough = true
	}

	if c.JobStatusAddr != "" {
	p.config.JobStatusAddr = c.JobStatusAddr
	}

	if c.PrototextWriterOptions.Path != "" {
	p.config.PrototextWriterOptions.Path = c.PrototextWriterOptions.Path
	}

	if c.StreamIngressOptions.InputFile != "" {
	p.config.SetFile(c.StreamIngressOptions.InputFile)
	} else if c.StreamIngressOptions.InputFIFO != "" {
	p.config.SetFIFO(c.StreamIngressOptions.InputFIFO)
	}
	if c.StreamIngressOptions.IsTemp {
	p.config.StreamIngressOptions.IsTemp = true
	}
	if c.StreamIngressOptions.OpenBlocking {
	p.config.StreamIngressOptions.OpenBlocking = true
	}
	if c.GRPCIngressOptions.Addr != "" {
	p.config.GRPCIngressOptions.Addr = c.GRPCIngressOptions.Addr
	}
	if c.PrototextReaderOptions.Path != "" {
	p.config.PrototextReaderOptions.Path = c.PrototextReaderOptions.Path
	}

	if c.IngressListener != nil {
	p.config.IngressListener = c.IngressListener
	}

	if c.RSConn != nil {
	p.config.RSConn = c.RSConn
	}
	if c.CASConn != nil {
	p.config.CASConn = c.CASConn
	}
	if c.JSConn != nil {
	p.config.JSConn = c.JSConn
	}
	if c.RPCTimeouts != nil {
	p.config.RPCTimeouts = c.RPCTimeouts
	}
	if c.Retrier != nil {
	p.config.Retrier = c.Retrier
	}
	}

	// ProxyPerRPCCredsOpt is an ProxyOpt that provides per-RPC credentials to the Proxy.
	type ProxyPerRPCCredsOpt struct {
	Creds credentials.PerRPCCredentials
	}

	// ApplyProxy implements ProxyOpt for ProxyPerRPCCredsOpt.
	func (p ProxyPerRPCCredsOpt) ApplyProxy(proxy Proxy) {
	proxy.creds = p.Creds
	}

	// DialParamsOpt is an ProxyOpt that provides dial parameters for ResultStore, CAS, and JobStatus.
	type DialParamsOpt struct {
	RSDialParams auth.DialParams
	CASDialParams auth.DialParams
	JSDialParams auth.DialParams
	}

	// ApplyProxy implements ProxyOpt for DialParamsOpt.
	func (d DialParamsOpt) ApplyProxy(p Proxy) {
	if d.RSDialParams.Service != "" {
	p.config.RSEgress = auth.NewServiceConnection(d.RSDialParams)
	}
	if d.CASDialParams.Service != "" {
	p.config.CASEgress = auth.NewServiceConnection(d.CASDialParams)
	}
	if d.JSDialParams.Service != "" {
	p.config.Conn = auth.NewServiceConnection(d.JSDialParams)
	}
	}

	// NewProxy creates a new Proxy and establishes connections to the ResultStore,
	// CAS, and (optionally) JobStatus services based on the provided configuration.
	func NewProxy(ctx context.Context, opts ...ProxyOpt) (*Proxy, error) {
	proxy := &Proxy{
	config: &ProxyConfig{
	ResultStoreEgressOptions: ResultStoreEgressOptions{
	BundleCountThreshold: DefaultBundleCountThreshold,
	BundleByteThreshold: DefaultBundleByteThreshold,
	},
	RPCTimeouts: DefaultRPCTimeouts,
	Retrier: RetryTransient(),
	},
	router: NewRouter(ctx),
	FinalizationSignal: make(chan struct{}),
	runStarted: make(chan struct{}),
	cleanupDone: make(chan bool),
	}

	// Phase 1: Apply all options to the configuration.
	for _, o := range opts {
	if o != nil {
	o.ApplyProxy(proxy)
	}
	}

	// Phase 2: Initialize workers based on the final configuration.
	// ResultStore Worker
	if w, err := proxy.config.ResultStoreEgressOptions.CreateWorker(ctx, proxy.config.RSConn, proxy.config.CASConn, proxy.config.Retrier, proxy.RPCOpts, proxy.FinalizationSignal); err != nil {
	return nil, err
	} else if w != nil {
	if err := proxy.router.RegisterEgressWorker(w); err != nil {
	proxy.Close(ctx)
	return nil, err
	}
	}

	// JobStatus Worker
	if w, err := proxy.config.JobStatusEgressOptions.CreateWorker(ctx, proxy.config.JSConn, proxy.config.Retrier, proxy.RPCOpts); err != nil {
	return nil, err
	} else if w != nil {
	if err := proxy.router.RegisterEgressWorker(w); err != nil {
	proxy.Close(ctx)
	return nil, err
	}
	}

	// PrototextWriter
	if w, err := proxy.config.PrototextWriterOptions.CreateWorker(); err != nil {
	return nil, err
	} else if w != nil {
	if err := proxy.router.RegisterEgressWorker(w); err != nil {
	proxy.Close(ctx)
	return nil, err
	}
	}

	// Phase 3: Initialize Ingress Worker
	if proxy.config.HasGRPCIngress() {
	lis := proxy.config.IngressListener
	if lis == nil && proxy.config.GRPCIngressOptions.Addr != "bufnet" {
	var err error
	lis, err = net.Listen("tcp", proxy.config.GRPCIngressOptions.Addr)
	if err != nil {
	return nil, err
	}
	proxy.config.GRPCIngressOptions.Addr = lis.Addr().String()
	}
	proxy.ingressWorker = NewGRPCIngressWorker(lis, proxy.signalIngressReady)
	} else if proxy.config.PrototextReaderOptions.Path != "" {
	proxy.ingressWorker = NewPrototextReader(proxy.config.PrototextReaderOptions.Path)
	} else if proxy.config.StreamIngressOptions.InputFIFO != "" \|\| proxy.config.StreamIngressOptions.InputFile != "" \|\| proxy.config.StreamIngressOptions.IsTemp {
	worker := NewStreamIngressWorker(proxy.config.StreamIngressOptions.ToInputSource(), proxy.signalIngressReady)

	// Setup the source (creates it on disk and potentially blocks on open).
	cleanup, err := worker.(*streamIngressWorker).Setup(ctx)
	if err != nil {
	proxy.Close(ctx)
	return nil, fmt.Errorf("failed to setup stream ingress: %w", err)
	}

	// Update the path in config (handles random temp names and input-type specific fields).
	path := worker.(*streamIngressWorker).Path()
	if proxy.config.StreamIngressOptions.InputFile != "" {
	proxy.config.StreamIngressOptions.InputFile = path
	} else {
	proxy.config.StreamIngressOptions.InputFIFO = path
	}
	proxy.ingressWorker = worker

	// Signal readiness immediately since the FIFO/File is now created on disk.
	proxy.signalIngressReady(IngressMetadata{FIFOPath: path})

	// Ensure the source is cleaned up when the proxy is closed.
	proxy.AddCleanup(cleanup)
	}

	return proxy, nil
	}

	// Close gracefully shuts down the proxy and all its registered workers.
	func (p *Proxy) Close(ctx context.Context) error {
	glog.Infof("proxy waiting for background tasks to finish")
	select {
	case <-p.runStarted:
	// Run was called, wait for the router to finish.
	default:
	// Run was never called, do nothing.
	return nil
	}
	glog.Infof("proxy background tasks finished")

	return p.router.Close(ctx)
	}

	// Run executes the internal Router, which kicks off all registered workers
	// in parallel. It returns when the ingress stream is finished and all workers
	// have successfully completed their processing.
	func (p *Proxy) Run(ctx context.Context) error {
	p.runStartedOnce.Do(func() {
	close(p.runStarted)
	})
	worker := p.ingressWorker
	if worker == nil {
	worker = noopIngressWorker{}
	p.signalIngressReady(IngressMetadata{})
	}
	if err := p.router.SetIngress(worker); err != nil {
	return err
	}
	return p.router.Run(ctx)
	}

	// Ready returns a channel that will receive IngressMetadata once the server
	// is configured and ready to process build events. This method is
	// thread-safe and supports multiple concurrent listeners (broadcast) as
	// well as late-joiners.
	func (p *Proxy) Ready() <-chan IngressMetadata {
	return p.ready.Ready()
	}

	// IngressDrainedChan returns a channel that is closed when the ingress
	// source has been completely consumed and drained by the dispatcher.
	func (p *Proxy) IngressDrainedChan() <-chan struct{} {
	return p.router.IngressDrainedChan()
	}

	// signalIngressReady signals that the proxy is ready to accept requests.
	func (p *Proxy) signalIngressReady(m IngressMetadata) {
	p.ready.Signal(m)
	}

	// SignalIngressEOF unblocks the proxy's ingress reader by sending a best-effort
	// EOF signal. This is primarily used for FIFO ingress to ensure the reader
	// loop doesn't hang if the build tool exits without ever opening the pipe.
	// It is a no-op if the ingress source is not a FIFO.
	func (p *Proxy) SignalIngressEOF() error {
	var m IngressMetadata
	if v, ok := p.ready.Value(); ok {
	m = v
	}
	if m.FIFOPath != "" {
	return fifo.SignalFIFOEOF(m.FIFOPath)
	}
	return nil
	}

	// SignalFinalization signals to all workers that the primary build phase is
	// complete and they should process post-build artifacts and finalize their
	// respective backend operations.
	func (p *Proxy) SignalFinalization() {
	p.fOnce.Do(func() {
	glog.Infof("Signaling workers to finalize invocation and upload post-build artifacts.")
	close(p.FinalizationSignal)
	})
	}

	// AddCleanup adds a function to be called during DrainAndReleaseResources.
	func (p *Proxy) AddCleanup(fn func()) {
	p.cleanupFns = append(p.cleanupFns, fn)
	}

	// DrainAndReleaseResources runs all registered cleanup functions.
	func (p *Proxy) DrainAndReleaseResources() {
	p.rrOnce.Do(func() {
	glog.Infof("Proxy shutting down.")
	defer p.ready.Close()
	defer func() {
	glog.Infof("Proxy shutdown complete.")
	close(p.cleanupDone)
	}()
	for _, cleanup := range p.cleanupFns {
	if cleanup != nil {
	cleanup()
	}
	}
	})
	}

	// Shutdown initiates a graceful shutdown of the proxy, cleaning up all resources.
	func (p *Proxy) Shutdown() {
	p.DrainAndReleaseResources()
	<-p.cleanupDone
	}

	// Router returns the internal Router instance managed by this proxy.
	func (p Proxy) Router() Router {
	return p.router
	}

	// Errors returns a snapshot of the non-fatal errors encountered
	// across all active workers.
	func (p *Proxy) Errors() []error {
	return p.router.Errors()
	}

	// HasEgress returns true if at least one egress worker is configured and ready to run.
	func (p *Proxy) HasEgress() bool {
	return p.router.HasEgress()
	}

	// HasGRPCIngress returns true if a gRPC ingress worker is configured.
	func (p *Proxy) HasGRPCIngress() bool {
	return p.config.HasGRPCIngress()
	}

	// RPCOpts returns the set of default gRPC call options (like authentication)
	// that should be applied to all requests made by this proxy.
	func (p *Proxy) RPCOpts() []grpc.CallOption {
	opts := []grpc.CallOption{}
	if p.creds == nil {
	return opts
	}
	return append(opts, grpc.PerRPCCredentials(p.creds))
	}

	// CallWithTimeout wraps a gRPC call with the configured timeout for that
	// specific operation.
	func (p *Proxy) CallWithTimeout(ctx context.Context, rpcName string, f func(ctx context.Context) error) error {
	timeout, ok := p.config.RPCTimeouts[rpcName]
	if !ok {
	if timeout, ok = p.config.RPCTimeouts["default"]; !ok {
	timeout = 0
	}
	}
	if timeout == 0 {
	return f(ctx)
	}
	childCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	return f(childCtx)
	}