internal/pkg/rsproxy/proxy_test.go

// 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"
	"net"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/bazelbuild/rsclient/internal/pkg/auth"
	"github.com/bazelbuild/rsclient/internal/pkg/fakejobstatus"
	"github.com/bazelbuild/rsclient/internal/pkg/fakeresultstore"
	"github.com/bazelbuild/rsclient/internal/pkg/fifo"
	"github.com/bazelbuild/rsclient/internal/pkg/fifo/fifotest"
	"github.com/bazelbuild/rsclient/internal/pkg/grpcutil"
	"github.com/bazelbuild/rsclient/internal/pkg/grpcutil/grpctest"
	"github.com/bazelbuild/rsclient/internal/pkg/netutil"
	"github.com/bazelbuild/rsclient/internal/pkg/rsmsg"
	"github.com/bazelbuild/rsclient/internal/pkg/testenv"
	"google.golang.org/grpc"
	"google.golang.org/grpc/test/bufconn"

	jspb "github.com/bazelbuild/rsclient/api/jobstatus"
	rspb "google.golang.org/genproto/googleapis/devtools/resultstore/v2"

	_ "github.com/bazelbuild/rsclient/internal/pkg/testsetup"
)

const (
	// pollTimeout is used for short-lived polling loops waiting for background
	// tasks to update shared state.
	pollTimeout = 2 * time.Second
)

// proxyTestEnv wraps the standard test environment with a Proxy instance
// configured for testing.
type proxyTestEnv struct {
	*testenv.Env
	proxy  *Proxy
	fakeRS *fakeresultstore.Server
	fakeJS *fakejobstatus.Server
}

// newProxyTestEnv initializes a fake backend environment and creates a Proxy
// connected to it via in-memory bufconn listeners.
func newProxyTestEnv(ctx context.Context, t *testing.T, opts ...ProxyOpt) (*proxyTestEnv, error) {
	// Use a sub-context with timeout for dialing and initialization of fake servers.
	setupCtx, cancel := context.WithTimeout(ctx, defaultTestTimeout)
	defer cancel()

	// Always provide standard test defaults.
	allOpts := []ProxyOpt{&ProxyConfig{
		ResultStoreEgressOptions: ResultStoreEgressOptions{
			PartialInvocation: &rspb.Invocation{
				InvocationAttributes: &rspb.InvocationAttributes{},
			},
			CASInstanceName: "instance",
		},
	}}
	allOpts = append(allOpts, opts...)

	env, err := testenv.New(setupCtx, t, testenv.WithResultStore(), testenv.WithJobStatus())
	if err != nil {
		return nil, err
	}
	// Use the fake connections from the test environment.
	params := ProxyEgressParams{
		RSConn:  env.RS.Conn,
		CASConn: env.CAS.Conn,
		JSConn:  env.JS.Conn,
	}
	// The Proxy itself should use the long-lived ctx.
	proxy, err := NewProxyFromConnection(ctx, params, allOpts...)
	if err != nil {
		return nil, err
	}

	// Register an automatic cleanup task to close the proxy.
	proxy.AddCleanup(func() {
		cleanupCtx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
		defer cancel()
		if err := proxy.Close(cleanupCtx); err != nil {
			if grpcutil.IsContextError(err) {
				return
			}
			t.Errorf("proxy.Close() failed: %v", err)
		}
	})

	return &proxyTestEnv{
		Env:    env,
		proxy:  proxy,
		fakeRS: env.RS.Impl,
		fakeJS: env.JS.Impl,
	}, nil
}

// TestProxyConfig_ApplyProxy_Merging verifies that different ProxyOpt types
// correctly merge their settings into the Proxy configuration without
// overwriting unrelated fields.
func TestProxyConfig_ApplyProxy_Merging(t *testing.T) {
	p := &Proxy{
		config: &ProxyConfig{
			ResultStoreEgressOptions: ResultStoreEgressOptions{
				BundleCountThreshold: 100,
				BundleByteThreshold:  1000,
			},
		},
	}

	// 1. Apply an option that only sets BundleCountThreshold.
	opt1 := &ProxyConfig{
		ResultStoreEgressOptions: ResultStoreEgressOptions{
			BundleCountThreshold: 50,
		},
	}
	opt1.ApplyProxy(p)

	if p.config.BundleCountThreshold != 50 {
		t.Errorf("BundleCountThreshold = %d, want 50", p.config.BundleCountThreshold)
	}
	if p.config.BundleByteThreshold != 1000 {
		t.Errorf("BundleByteThreshold = %d, want 1000 (preserved)", p.config.BundleByteThreshold)
	}

	// 2. Apply an option that sets RSDialParams via DialParamsOpt.
	opt2 := &DialParamsOpt{
		RSDialParams: auth.DialParams{Service: "myservice"},
	}
	opt2.ApplyProxy(p)

	if p.config.RSEgress == nil || p.config.RSEgress.DialParams.Service != "myservice" {
		t.Errorf("RSEgress.Service = %v, want \"myservice\"", p.config.RSEgress)
	}
	if p.config.BundleCountThreshold != 50 {
		t.Errorf("BundleCountThreshold = %d, want 50 (preserved)", p.config.BundleCountThreshold)
	}
}

// TestProxy_Run verifies the core build event fan-out pipeline of the Proxy.
func TestProxy_Run(t *testing.T) {
	testCases := []struct {
		name     string
		requests []*rspb.UploadRequest
	}{
		{
			name:     "empty request stream",
			requests: nil,
		},
		{
			name: "single create request",
			requests: []*rspb.UploadRequest{
				rsmsg.NewInvocationUploadRequest(rspb.UploadRequest_CREATE, rsmsg.NewInvocation("test-invocation", rsmsg.StatusUnspecified)),
			},
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			testCtx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
			defer cancel()

			fifoPath := fifotest.NewTestFIFO(t)

			opts := []ProxyOpt{&ProxyConfig{
				StreamIngressOptions: StreamIngressOptions{
					InputFIFO: fifoPath,
				},
			}}

			env, err := newProxyTestEnv(testCtx, t, opts...)
			if err != nil {
				t.Fatalf("newProxyTestEnv() failed in test case %q: %v", tc.name, err)
			}
			defer env.proxy.Shutdown()

			// Ground the FIFO with KeepOpen to prevent premature reader EOF.
			ko := fifotest.NewKeepOpen(t, fifoPath)
			defer ko.Close()

			var wg sync.WaitGroup
			wg.Add(2)

			go func() {
				defer wg.Done()
				if err := env.proxy.Run(testCtx); err != nil && err != context.Canceled {
					t.Errorf("Proxy.Run() failed in test case %q: %v", tc.name, err)
				}
			}()

			// Write requests to unblock the proxy.
			go func() {
				defer wg.Done()
				// Open the writer and hand off from KeepOpen.
				fw, err := ko.OpenWriter(t.Context())
				if err != nil {
					t.Errorf("failed to open FIFO for writing in test case %q: %v", tc.name, err)
					return
				}
				defer fw.Close()

				for _, req := range tc.requests {
					if err := WriteUploadRequest(fw, req); err != nil {
						t.Errorf("failed to write upload request in test case %q: %v", tc.name, err)
						return
					}
				}
				// After writing all requests, signal the proxy to finalize.
				env.proxy.SignalFinalization()
			}()

			wg.Wait()
		})
	}
}

// TestProxy_ReadySignal verifies that the Proxy signals readiness on Ready()
// as soon as its internal ingress processing is initialized and ready for traffic.
func TestProxy_ReadySignal(t *testing.T) {
	testCtx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	fifoPath := fifotest.NewTestFIFO(t)

	opts := []ProxyOpt{&ProxyConfig{
		StreamIngressOptions: StreamIngressOptions{
			InputFIFO: fifoPath,
		},
	}}

	env, err := newProxyTestEnv(testCtx, t, opts...)
	if err != nil {
		t.Fatalf("newProxyTestEnv() failed: %v", err)
	}
	defer env.proxy.Shutdown()

	// Ground the FIFO with KeepOpen to prevent premature reader EOF.
	ko := fifotest.NewKeepOpen(t, fifoPath)
	defer ko.Close()

	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		env.proxy.Run(testCtx)
	}()

	select {
	case <-env.proxy.Ready():
		// Success.
	case <-testCtx.Done():
		t.Fatal("Timeout waiting for ready signal")
	}

	// Unblock Run() by closing the writer end of the FIFO.
	go func() {
		defer wg.Done()
		fw, err := ko.OpenWriter(t.Context())
		if err != nil {
			t.Errorf("failed to open FIFO for writing: %v", err)
			return
		}
		fw.Close()
	}()
	wg.Wait()
}

// TestProxy_ReadyMetadata verifies that the metadata returned by Ready()
// contains the correct information for different ingress types.
func TestProxy_ReadyMetadata(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	t.Run("FIFO", func(t *testing.T) {
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()

		fifoPath := fifotest.NewTestFIFO(t)
		p, err := NewProxy(ctx, &ProxyConfig{
			StreamIngressOptions: StreamIngressOptions{
				InputFIFO: fifoPath,
			},
		})
		if err != nil {
			t.Fatalf("NewProxy failed: %v", err)
		}
		defer p.Close(ctx)

		go p.Run(ctx)

		select {
		case m := <-p.Ready():
			if m.FIFOPath != fifoPath {
				t.Errorf("FIFOPath = %q, want %q", m.FIFOPath, fifoPath)
			}
			if m.GRPCAddr != "" {
				t.Errorf("GRPCAddr = %q, want empty", m.GRPCAddr)
			}
		case <-ctx.Done():
			t.Fatal("Timeout waiting for ready signal")
		}
	})

	t.Run("GRPC", func(t *testing.T) {
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()

		lis, err := net.Listen("tcp", netutil.DefaultLocalIPv4Addr)
		if err != nil {
			t.Fatalf("net.Listen failed: %v", err)
		}
		addr := lis.Addr().String()

		p, err := NewProxy(ctx, &ProxyConfig{
			IngressListener: lis,
		})
		if err != nil {
			t.Fatalf("NewProxy failed: %v", err)
		}
		defer p.Close(ctx)

		go p.Run(ctx)

		select {
		case m := <-p.Ready():
			if m.GRPCAddr != addr {
				t.Errorf("GRPCAddr = %q, want %q", m.GRPCAddr, addr)
			}
			if m.FIFOPath != "" {
				t.Errorf("FIFOPath = %q, want empty", m.FIFOPath)
			}
		case <-ctx.Done():
			t.Fatal("Timeout waiting for ready signal")
		}
	})
}

// TestProxy_SignalFinalization verifies that SignalFinalization() correctly
// notifies all internal workers to process post-build artifacts.
func TestProxy_SignalFinalization(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	env, err := newProxyTestEnv(ctx, t)
	if err != nil {
		t.Fatalf("newProxyTestEnv() failed: %v", err)
	}
	defer env.proxy.Shutdown()

	env.proxy.SignalFinalization()

	select {
	case <-env.proxy.FinalizationSignal:
		// Success.
	case <-time.After(1 * time.Second):
		t.Fatal("Timeout waiting for finalization signal")
	}
}

// TestProxy_AddCleanup verifies that registered cleanup functions are executed
// during Proxy shutdown.
func TestProxy_AddCleanup(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	env, err := newProxyTestEnv(ctx, t)
	if err != nil {
		t.Fatalf("newProxyTestEnv() failed: %v", err)
	}

	cleanupCalled := false
	env.proxy.AddCleanup(func() {
		cleanupCalled = true
	})

	env.proxy.Shutdown()

	if !cleanupCalled {
		t.Error("Cleanup function was not called during shutdown")
	}
}

// TestProxy_IngressDrainedSignal verifies that IngressDrainedChan is closed
// only after the ingress worker has finished and the dispatcher has emptied
// the inbound queue.
func TestProxy_IngressDrainedSignal(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	fifoPath := fifotest.NewTestFIFO(t)

	// In Sink Mode, we don't need newProxyTestEnv's fakes.
	// We use OpenBlocking=true to ensure precise synchronization.
	opts := []ProxyOpt{
		&ProxyConfig{
			StreamIngressOptions: StreamIngressOptions{
				InputFIFO:    fifoPath,
				OpenBlocking: true,
			},
		},
		&DialParamsOpt{}, // Effectively disables standard egress.
	}

	var wg sync.WaitGroup
	wg.Add(1)

	writerClosed := make(chan struct{})
	canCloseWriter := make(chan struct{})

	// 1. Start background writer BEFORE creating the Proxy.
	// NewProxy will call Setup(), which blocks until this writer connects.
	req := rsmsg.NewInvocationUploadRequest(rspb.UploadRequest_CREATE, rsmsg.NewInvocation("drain-test", rsmsg.StatusUnspecified))
	go func() {
		defer wg.Done()
		fw, err := fifo.OpenWriterContext(t.Context(), fifoPath)
		if err != nil {
			t.Errorf("failed to open FIFO for writing: %v", err)
			return
		}
		if err := WriteUploadRequest(fw, req); err != nil {
			t.Errorf("failed to write upload request: %v", err)
			fw.Close()
			return
		}

		// Keep the writer end open until the test signals we can close it.
		<-canCloseWriter
		fw.Close()
		close(writerClosed)
	}()

	p, err := NewProxy(ctx, opts...)
	if err != nil {
		t.Fatalf("NewProxy() failed: %v", err)
	}
	defer p.Close(ctx)

	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := p.Run(ctx); err != nil && err != context.Canceled {
			t.Errorf("Proxy.Run() failed: %v", err)
		}
	}()

	// Wait for the proxy to be ready.
	select {
	case <-p.Ready():
	case <-ctx.Done():
		t.Fatal("Timeout waiting for proxy to be ready")
	}

	// 2. The drained channel should still be open while the writer is connected.
	select {
	case <-p.IngressDrainedChan():
		t.Fatal("IngressDrainedChan closed prematurely while writer was connected")
	default:
		// Success.
	}

	// 3. Now allow the writer to close, which should eventually close IngressDrainedChan.
	close(canCloseWriter)
	<-writerClosed

	select {
	case <-p.IngressDrainedChan():
		// Success.
	case <-ctx.Done():
		t.Errorf("Timeout waiting for ingress drained signal after writer closed")
	}
}

// TestProxy_MultiEgress verifies that the Proxy correctly initializes and
// orchestrates multiple egress workers simultaneously, ensuring that incoming
// events are fanned out to all of them.
func TestProxy_MultiEgress(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	tmpDir := t.TempDir()
	dumpPath := filepath.Join(tmpDir, "dump.txt")

	fifoPath := fifotest.NewTestFIFO(t)

	// 1. Configure Proxy with multiple egresses: ResultStore and PrototextWriter.
	opts := []ProxyOpt{&ProxyConfig{
		ResultStoreEgressOptions: ResultStoreEgressOptions{
			RSInstanceName:  "projects/test/instances/default",
			CASInstanceName: "projects/test/instances/default",
		},
		PrototextWriterOptions: PrototextWriterOptions{
			Path: dumpPath,
		},
		StreamIngressOptions: StreamIngressOptions{
			InputFIFO: fifoPath,
		},
	}}

	env, err := newProxyTestEnv(ctx, t, opts...)
	if err != nil {
		t.Fatalf("newProxyTestEnv() failed: %v", err)
	}
	defer env.proxy.Shutdown()

	// Ground the FIFO with KeepOpen to prevent premature reader EOF.
	ko := fifotest.NewKeepOpen(t, fifoPath)
	defer ko.Close()

	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		if err := env.proxy.Run(ctx); err != nil && err != context.Canceled {
			t.Errorf("Proxy.Run() failed: %v", err)
		}
	}()

	// 2. Send build events.
	req1 := rsmsg.NewInvocationUploadRequest(rspb.UploadRequest_CREATE,
		rsmsg.NewInvocation("test-multi-egress", rsmsg.StatusUnspecified))

	go func() {
		defer wg.Done()
		fw, err := ko.OpenWriter(t.Context())
		if err != nil {
			t.Errorf("failed to open FIFO for writing: %v", err)
			return
		}
		defer fw.Close()

		if err := WriteUploadRequest(fw, req1); err != nil {
			t.Errorf("failed to write req1: %v", err)
		}
		env.proxy.SignalFinalization()
	}()

	// Wait for the proxy and writer goroutines to finish.
	// Router.Run() ensures that all egress workers have finished their loops.
	wg.Wait()

	// 3. Verify ResultStore egress (via FakeResultStore).
	invName := "invocations/test-multi-egress"
	err = env.fakeRS.WaitUntil(ctx, func(data fakeresultstore.ServerData) bool {
		return data.FinalState.Invocations[invName] != nil
	})

	if err != nil {
		t.Errorf("Invocation %q was not created in ResultStore: %v", invName, err)
		// Print all recorded errors to help diagnose flakiness.
		for _, err := range env.proxy.Errors() {
			t.Logf("Recorded error: %v", err)
		}
	}

	// 4. Verify PrototextWriter egress (via dump file).
	var dumpContent []byte
	start := time.Now()
	for time.Since(start) < fillTimeout {
		dumpContent, _ = os.ReadFile(dumpPath)
		if len(dumpContent) > 0 {
			break
		}
		time.Sleep(pollInterval)
	}

	if len(dumpContent) == 0 {
		t.Error("Dump file is empty; PrototextWriter failed to record requests")
	}
}

// TestProxy_AutonomousFIFOIngress verifies that the Proxy can autonomously
// manage its own FIFO ingress (creation and cleanup).
func TestProxy_AutonomousFIFOIngress(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	// 1. Configure Proxy for autonomous temporary FIFO with non-blocking open.
	// This matches the integrated mode configuration.
	opts := []ProxyOpt{&ProxyConfig{
		StreamIngressOptions: StreamIngressOptions{
			IsTemp:       true,
			OpenBlocking: false,
		},
	}}

	env, err := newProxyTestEnv(ctx, t, opts...)
	if err != nil {
		t.Fatalf("newProxyTestEnv() failed: %v", err)
	}

	// 3. Open the FIFO for writing using the Patient Writer.
	// This ensures the writer end is connected before the proxy starts reading,
	// preventing a premature EOF if the proxy starts faster than the writer.
	m := <-env.proxy.Ready()
	fifoPath := m.FIFOPath
	if fifoPath == "" {
		t.Fatal("Proxy.Ready() returned empty FIFOPath for autonomous FIFO")
	}

	// 2. Verify FIFO exists on disk.
	if _, err := os.Stat(fifoPath); os.IsNotExist(err) {
		t.Errorf("FIFO file %s was not created", fifoPath)
	}

	fw, err := fifo.OpenWriterContext(t.Context(), fifoPath)

	if err != nil {
		t.Fatalf("Failed to open FIFO for writing: %v", err)
	}
	defer fw.Close()

	// 4. Start the proxy.
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := env.proxy.Run(ctx); err != nil && err != context.Canceled {
			t.Errorf("Proxy.Run() failed: %v", err)
		}
	}()

	// 5. Send requests to the FIFO.
	req := rsmsg.NewInvocationUploadRequest(rspb.UploadRequest_CREATE, rsmsg.NewInvocation("fifo-test", rsmsg.StatusUnspecified))
	if err := WriteUploadRequest(fw, req); err != nil {
		t.Errorf("Failed to write to FIFO: %v", err)
	}

	// 6. Finalize and verify data reached backend.
	// We explicitly signal finalization then close the writer to trigger EOF.
	env.proxy.SignalFinalization()
	fw.Close()
	wg.Wait()

	invName := "invocations/fifo-test"
	gotInvoc := env.fakeRS.Data().FinalState.Invocations[invName]
	if gotInvoc == nil {
		t.Errorf("ResultStore: Invocation %q not received via autonomous FIFO", invName)
	}

	// 7. Shutdown and verify cleanup.
	env.proxy.Shutdown()
	if _, err := os.Stat(fifoPath); err == nil {
		t.Errorf("FIFO file %s was not removed after shutdown", fifoPath)
	}
}

// TestNewProxy_SinkMode verifies that the Proxy can be initialized without
// any egress backend services, in which case it operates in "sink" mode
// (consuming events but doing nothing with them).
func TestNewProxy_SinkMode(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	// NewProxy should succeed even if no services are specified.
	proxy, err := NewProxy(ctx, &DialParamsOpt{
		RSDialParams:  auth.DialParams{},
		CASDialParams: auth.DialParams{},
	})
	if err != nil {
		t.Fatalf("NewProxy() with no services failed: %v", err)
	}
	if proxy == nil {
		t.Fatal("NewProxy() returned nil proxy")
	}
	defer proxy.Close(ctx)

	if proxy.HasEgress() {
		t.Error("proxy.HasEgress() expected false (Sink mode), but got true")
	}

	// NewProxy should succeed if at least JobStatus is specified, even if
	// ResultStore and CAS are empty.
	jsServer := fakejobstatus.New()
	jsBackend := grpctest.NewTCPServer(t, func(s *grpc.Server) {
		jspb.RegisterJobStatusServer(s, jsServer)
	})

	opts := []ProxyOpt{&ProxyConfig{
		ResultStoreEgressOptions: ResultStoreEgressOptions{
			CASInstanceName: "instance",
		},
		JobStatusEgressOptions: JobStatusEgressOptions{
			JobStatusAddr: jsBackend.Addr(),
		},
	}}
	proxyJS, err := NewProxy(ctx, append(opts, &DialParamsOpt{
		JSDialParams: auth.DialParams{
			Service:    jsBackend.Addr(),
			NoSecurity: true,
		},
	})...)
	if err != nil {
		t.Fatalf("NewProxy() with JobStatusAddr failed: %v", err)
	}
	if proxyJS == nil {
		t.Fatal("NewProxy() returned nil proxy")
	}
	defer proxyJS.Close(ctx)

	if !proxyJS.HasEgress() {
		t.Error("proxyJS.HasEgress() expected true (JobStatus), but got false")
	}
}

// TestProxy_AutonomousGRPCIngress verifies that the Proxy can manage its own
// gRPC ingress server when a listener is provided.
func TestProxy_AutonomousGRPCIngress(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), extendedTestTimeout)
	defer cancel()

	lis := bufconn.Listen(1024 * 1024)

	// 1. Configure Proxy with gRPC ingress via bufconn.
	opts := []ProxyOpt{&ProxyConfig{
		IngressListener: lis,
	}}

	env, err := newProxyTestEnv(ctx, t, opts...)
	if err != nil {
		t.Fatalf("newProxyTestEnv() failed: %v", err)
	}

	var wg sync.WaitGroup
	wg.Add(1)
	proxyCtx, proxyCancel := context.WithCancel(ctx)
	go func() {
		defer wg.Done()
		if err := env.proxy.Run(proxyCtx); err != nil && err != context.Canceled {
			t.Errorf("Proxy.Run() failed: %v", err)
		}
	}()

	// 2. Wait for the proxy (and its internal gRPC server) to be ready.
	select {
	case <-env.proxy.Ready():
	case <-ctx.Done():
		t.Fatal("Timeout waiting for proxy to be ready")
	}

	// 3. Connect to the autonomous ingress and send a request.
	conn := grpctest.ConnectListener(ctx, t, lis)
	defer conn.Close()

	client := rspb.NewResultStoreUploadClient(conn)
	invocID := "autonomous-grpc-test"
	_, err = client.CreateInvocation(ctx, &rspb.CreateInvocationRequest{
		InvocationId: invocID,
		Invocation: &rspb.Invocation{
			Id: &rspb.Invocation_Id{InvocationId: invocID},
		},
	})
	if err != nil {
		t.Fatalf("CreateInvocation failed: %v", err)
	}

	// 4. Verify request reached the fake backend.
	err = env.fakeRS.WaitUntil(ctx, func(data fakeresultstore.ServerData) bool {
		return len(data.CreateInvReqs) > 0
	})

	if err != nil {
		t.Errorf("ResultStore: CreateInvocation not received via autonomous gRPC ingress: %v", err)
	}

	// 5. Finalize the session via gRPC.
	_, err = client.FinalizeInvocation(ctx, &rspb.FinalizeInvocationRequest{
		Name: "invocations/" + invocID,
	})
	if err != nil {
		t.Fatalf("FinalizeInvocation failed: %v", err)
	}

	// 6. Cleanup: Close client connection first to unblock GracefulStop!
	conn.Close()

	// 7. Stop proxy.
	proxyCancel()
	wg.Wait()
}

// TestProxy_CloseUnstartedProxy verifies that Close() returns immediately and
// does not deadlock if Run() was never called.
func TestProxy_CloseUnstartedProxy(t *testing.T) {
	ctx, cancel := context.WithTimeout(t.Context(), shortTestTimeout)
	defer cancel()

	env, err := newProxyTestEnv(ctx, t)
	if err != nil {
		t.Fatalf("newProxyTestEnv() failed: %v", err)
	}

	closeDone := make(chan error, 1)
	go func() {
		closeDone <- env.proxy.Close(ctx)
	}()

	select {
	case err := <-closeDone:
		if err != nil {
			t.Errorf("Close() failed: %v", err)
		}
	case <-ctx.Done():
		t.Fatal("Close() deadlocked on unstarted proxy")
	}
}

// TestProxy_SignalIngressEOF verifies that the Proxy can unblock its ingress
// reader by signaling EOF to its FIFO.
func TestProxy_SignalIngressEOF(t *testing.T) {
	ctx := t.Context()
	fifoPath := fifotest.NewTestFIFO(t)

	opts := []ProxyOpt{
		&ProxyConfig{
			StreamIngressOptions: StreamIngressOptions{
				InputFIFO:    fifoPath,
				OpenBlocking: false,
			},
		},
	}
	p, err := NewProxy(ctx, opts...)
	if err != nil {
		t.Fatalf("NewProxy failed: %v", err)
	}
	defer p.Close(ctx)

	if err := p.SignalIngressEOF(); err != nil {
		t.Errorf("SignalIngressEOF failed: %v", err)
	}
}

// TestProxy_HasEgress verifies that the Proxy correctly reports whether it
// has any registered egress workers.
func TestProxy_HasEgress(t *testing.T) {
	ctx := t.Context()

	// Verify that a Proxy with no workers reports false.
	t.Run("NoEgress", func(t *testing.T) {
		p, _ := NewProxy(ctx)
		if p.HasEgress() {
			t.Error("HasEgress() = true, want false")
		}
	})

	// Verify that a Proxy with a configured Prototext worker reports true.
	t.Run("WithEgress", func(t *testing.T) {
		p, _ := NewProxy(ctx, &ProxyConfig{
			PrototextWriterOptions: PrototextWriterOptions{
				Path: filepath.Join(t.TempDir(), "dump.txt"),
			},
		})
		if !p.HasEgress() {
			t.Error("HasEgress() = false, want true")
		}
	})
}