From 2151d1e93c8dec3b81e337c314cc53a83590c813 Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Mon, 17 Aug 2026 11:10:19 -0500 Subject: [PATCH 1/8] atunnel: ingress CONNECT support Signed-off-by: Keith Mattix II --- .../internal/controllers/workerpool_apply.go | 7 +- .../controllers/workerpool_apply_test.go | 7 +- cmd/ateom-gvisor/main.go | 23 ++- cmd/ateom-microvm/main.go | 23 ++- demos/counter/counter.go | 23 +++ demos/counter/counter.yaml.tmpl | 1 + internal/atunnel/ingress.go | 176 +++++++++++++++--- internal/atunnel/ingress_test.go | 38 ++++ 8 files changed, 263 insertions(+), 35 deletions(-) diff --git a/cmd/atecontroller/internal/controllers/workerpool_apply.go b/cmd/atecontroller/internal/controllers/workerpool_apply.go index 0128ab3d4..1153605e3 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_apply.go +++ b/cmd/atecontroller/internal/controllers/workerpool_apply.go @@ -78,6 +78,7 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin WithArgs( "--pod-uid=$(POD_UID)", "--atunnel-listen-address=0.0.0.0:443", + "--atunnel-connect-listen-address=0.0.0.0:444", "--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem", "--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem", "--atunnel-egress-listen-address=0.0.0.0:15001", @@ -86,7 +87,11 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin WithPorts(corev1ac.ContainerPort(). WithName("https"). WithContainerPort(443). - WithProtocol(corev1.ProtocolTCP)). + WithProtocol(corev1.ProtocolTCP), + corev1ac.ContainerPort(). + WithName("connect"). + WithContainerPort(444). + WithProtocol(corev1.ProtocolTCP)). WithSecurityContext(ateomSecurityContext(wp.Spec.SandboxClass)). WithEnv(ateomContainerEnv(otel)...). WithVolumeMounts( diff --git a/cmd/atecontroller/internal/controllers/workerpool_apply_test.go b/cmd/atecontroller/internal/controllers/workerpool_apply_test.go index 9e73c6d73..02310337e 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_apply_test.go +++ b/cmd/atecontroller/internal/controllers/workerpool_apply_test.go @@ -735,6 +735,7 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf WithArgs( "--pod-uid=$(POD_UID)", "--atunnel-listen-address=0.0.0.0:443", + "--atunnel-connect-listen-address=0.0.0.0:444", "--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem", "--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem", "--atunnel-egress-listen-address=0.0.0.0:15001", @@ -743,7 +744,11 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf WithPorts(corev1ac.ContainerPort(). WithName("https"). WithContainerPort(443). - WithProtocol(corev1.ProtocolTCP)). + WithProtocol(corev1.ProtocolTCP), + corev1ac.ContainerPort(). + WithName("connect"). + WithContainerPort(444). + WithProtocol(corev1.ProtocolTCP)). WithSecurityContext(corev1ac.SecurityContext(). WithRunAsUser(0). WithRunAsGroup(0). diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index dc59c6fd8..ce165ca18 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -65,12 +65,13 @@ var ( podUID = pflag.String("pod-uid", "", "The UID of the current pod") // TODO(liorlieberman) have a sub package for all atunnel releated things like that - atunnelListenAddress = pflag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") - workerCredentialBundle = pflag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") - podIdentityTrustBundle = pflag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") - atunnelClientIdentity = pflag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") - atunnelEgressListenAddress = pflag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") - egressGatewayTrustBundle = pflag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") + atunnelListenAddress = pflag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") + atunnelConnectListenAddress = pflag.String("atunnel-connect-listen-address", "0.0.0.0:444", "Address for actor ingress mTLS CONNECT") + workerCredentialBundle = pflag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") + podIdentityTrustBundle = pflag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") + atunnelClientIdentity = pflag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") + atunnelEgressListenAddress = pflag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") + egressGatewayTrustBundle = pflag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") showVersion = pflag.Bool("version", false, "Print version and exit.") logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.") @@ -261,6 +262,16 @@ func runAtunnel(ctx context.Context, upstream *url.URL) (*atunnel.Server, *atunn } }() slog.InfoContext(ctx, "atunnel serving", slog.String("address", *atunnelListenAddress)) + atunnelConnectListener, err := net.Listen("tcp", *atunnelConnectListenAddress) + if err != nil { + return nil, nil, 0, fmt.Errorf("while opening atunnel CONNECT listener: %w", err) + } + go func() { + if err := atunnelIngress.ServeConnect(ctx, atunnelConnectListener); err != nil { + serverboot.Fatal(ctx, "Failed to serve actor CONNECT ingress", err) + } + }() + slog.InfoContext(ctx, "atunnel CONNECT serving", slog.String("address", *atunnelConnectListenAddress)) atunnelEgress, err := atunnel.NewEgress(atunnel.TCPOriginalDestination) if err != nil { diff --git a/cmd/ateom-microvm/main.go b/cmd/ateom-microvm/main.go index f94b2eb1a..af9c4c443 100644 --- a/cmd/ateom-microvm/main.go +++ b/cmd/ateom-microvm/main.go @@ -71,12 +71,13 @@ var ( otlpRelaySocket = flag.String("otlp-relay-socket", ateompath.AteletOTLPSocketPath(), "Unix socket of atelet's OTLP relay to export telemetry through, keeping it off the pod network. Empty, or absent at startup, exports directly to OTEL_EXPORTER_OTLP_ENDPOINT instead.") - atunnelListenAddress = flag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") - workerCredentialBundle = flag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") - podIdentityTrustBundle = flag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") - atunnelClientIdentity = flag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") - atunnelEgressListenAddress = flag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") - egressGatewayTrustBundle = flag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") + atunnelListenAddress = flag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") + atunnelConnectListenAddress = flag.String("atunnel-connect-listen-address", "0.0.0.0:444", "Address for actor ingress mTLS CONNECT") + workerCredentialBundle = flag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") + podIdentityTrustBundle = flag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") + atunnelClientIdentity = flag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS") + atunnelEgressListenAddress = flag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP") + egressGatewayTrustBundle = flag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway") ) const ( @@ -221,6 +222,16 @@ func do(ctx context.Context) error { } }() slog.InfoContext(ctx, "atunnel serving", slog.String("address", *atunnelListenAddress)) + atunnelConnectListener, err := net.Listen("tcp", *atunnelConnectListenAddress) + if err != nil { + return fmt.Errorf("while opening atunnel CONNECT listener: %w", err) + } + go func() { + if err := atunnelIngress.ServeConnect(ctx, atunnelConnectListener); err != nil { + serverboot.Fatal(ctx, "Failed to serve actor CONNECT ingress", err) + } + }() + slog.InfoContext(ctx, "atunnel CONNECT serving", slog.String("address", *atunnelConnectListenAddress)) atunnelEgress, err := atunnel.NewEgress(atunnel.TCPOriginalDestination) if err != nil { return fmt.Errorf("while configuring atunnel egress: %w", err) diff --git a/demos/counter/counter.go b/demos/counter/counter.go index d5e9ad637..a3927a6dc 100644 --- a/demos/counter/counter.go +++ b/demos/counter/counter.go @@ -69,6 +69,7 @@ func main() { secondFileCounterDirectory := pflag.String("second-file-counter-directory", "", "Directory for a second file counter; empty disables it. Used to exercise an Actor with more than one durable volume") validateExistingFilePath := pflag.String("validate-existing-file-path", "", "Path to existing file to validate reading") extraPort := pflag.Int("extra-port", 0, "Additional port to listen on, for exercising atenet-router's arbitrary-port ingress support; 0 disables it") + tcpPort := pflag.Int("tcp-port", 0, "Plain TCP echo port for exercising atunnel CONNECT ingress; 0 disables it") pflag.Parse() ctx := context.Background() @@ -177,6 +178,28 @@ func main() { }() } + if *tcpPort > 0 { + go func() { + listener, err := net.Listen("tcp", fmt.Sprintf(":%d", *tcpPort)) + if err != nil { + slog.ErrorContext(ctx, "Error starting counter TCP echo server", slog.Any("err", err)) + os.Exit(1) + } + slog.InfoContext(ctx, "Starting counter TCP echo server", slog.Int("port", *tcpPort)) + for { + conn, err := listener.Accept() + if err != nil { + slog.ErrorContext(ctx, "Counter TCP echo accept failed", slog.Any("err", err)) + return + } + go func() { + defer conn.Close() + _, _ = io.Copy(conn, conn) + }() + } + }() + } + // Write some random data to a file in the root filesystem, to test // filesystem checkpoint/restore. if err := writeRandomFile(); err != nil { diff --git a/demos/counter/counter.yaml.tmpl b/demos/counter/counter.yaml.tmpl index 10b60ff6a..db92de4fe 100644 --- a/demos/counter/counter.yaml.tmpl +++ b/demos/counter/counter.yaml.tmpl @@ -69,6 +69,7 @@ spec: # listener a test can address by CONNECTing to :9090, distinct # from the primary port 80 every other assertion in this demo uses. - --extra-port=9090 + - --tcp-port=9091 ${VALIDATE_EXISTING_FILE_PATH_ARG} readyz: httpGet: diff --git a/internal/atunnel/ingress.go b/internal/atunnel/ingress.go index f00d2bb7b..51d3788bc 100644 --- a/internal/atunnel/ingress.go +++ b/internal/atunnel/ingress.go @@ -21,6 +21,7 @@ import ( "crypto/x509" "errors" "fmt" + "io" "log/slog" "net" "net/http" @@ -36,6 +37,10 @@ import ( ) const ( + // DefaultConnectPort is the worker port on which atunnel accepts inbound + // mTLS CONNECT tunnels from the ingress router. + DefaultConnectPort = 444 + // StaleAssignmentHeader distinguishes an atunnel routing rejection from a // 421 returned by the actor application itself. StaleAssignmentHeader = "X-Ate-Assignment-Stale" @@ -78,6 +83,7 @@ type Server struct { credentialBundlePath string tlsConfig *tls.Config proxy *httputil.ReverseProxy + upstream *url.URL mu sync.Mutex active *activation @@ -146,9 +152,14 @@ func NewServer(cfg Config) (*Server, error) { s := &Server{ credentialBundlePath: cfg.CredentialBundlePath, proxy: proxy, + upstream: cfg.Upstream, } s.tlsConfig = &tls.Config{ MinVersion: tls.VersionTLS12, + // HTTP/1.1 CONNECT is relayed by hijacking the connection; HTTP/2 + // CONNECT uses its request and response streams instead. Advertising both + // makes the choice an ALPN negotiation rather than a listener setting. + NextProtos: []string{"h2", "http/1.1"}, GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) { return loadCredentialBundle(s.credentialBundlePath) }, @@ -188,8 +199,19 @@ func loadCredentialBundle(path string) (*tls.Certificate, error) { // Serve serves HTTPS on lis until ctx is canceled or the server fails. func (s *Server) Serve(ctx context.Context, lis net.Listener) error { + return s.serve(ctx, lis, s) +} + +// ServeConnect serves the mTLS CONNECT endpoint. CONNECT is deliberately on a +// separate listener so ordinary actor ingress remains a request proxy, while +// the router can use this listener for a bidirectional tunnel. +func (s *Server) ServeConnect(ctx context.Context, lis net.Listener) error { + return s.serve(ctx, lis, http.HandlerFunc(s.ServeConnectHTTP)) +} + +func (s *Server) serve(ctx context.Context, lis net.Listener, handler http.Handler) error { httpServer := &http.Server{ - Handler: s, + Handler: handler, TLSConfig: s.tlsConfig, ReadHeaderTimeout: 10 * time.Second, } @@ -209,6 +231,108 @@ func (s *Server) Serve(ctx context.Context, lis net.Listener) error { return err } +// ServeConnectHTTP accepts a router-authenticated CONNECT request and relays +// its tunnel to the named port on the currently active actor. +func (s *Server) ServeConnectHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodConnect { + http.Error(w, "CONNECT required", http.StatusMethodNotAllowed) + return + } + ref, ctx, release, ok := s.authorize(r) + if !ok { + s.reject(w) + return + } + defer release() + + _, port, err := net.SplitHostPort(r.Host) + if err != nil { + http.Error(w, "CONNECT authority must include a port", http.StatusBadRequest) + return + } + if _, ok := ParsePort(port); !ok { + http.Error(w, "invalid CONNECT port", http.StatusBadRequest) + return + } + + upstream, err := net.DialTimeout("tcp", net.JoinHostPort(s.upstream.Hostname(), port), 5*time.Second) + if err != nil { + slog.WarnContext(r.Context(), "atunnel CONNECT upstream failed", slog.Any("actor", ref), slog.Any("err", err)) + http.Error(w, "bad gateway", http.StatusBadGateway) + return + } + defer upstream.Close() + + if r.ProtoMajor == 2 { + s.serveH2Connect(w, r, upstream, ctx) + return + } + s.serveH1Connect(w, upstream, ctx) +} + +func (s *Server) serveH1Connect(w http.ResponseWriter, upstream net.Conn, ctx context.Context) { + hj, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "CONNECT hijacking unsupported", http.StatusInternalServerError) + return + } + client, rw, err := hj.Hijack() + if err != nil { + return + } + defer client.Close() + if _, err := rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n"); err != nil { + return + } + if err := rw.Flush(); err != nil { + return + } + + done := make(chan struct{}, 2) + go func() { _, _ = io.Copy(upstream, rw); done <- struct{}{} }() + go func() { _, _ = io.Copy(client, upstream); done <- struct{}{} }() + select { + case <-ctx.Done(): + case <-done: + } +} + +func (s *Server) serveH2Connect(w http.ResponseWriter, r *http.Request, upstream net.Conn, ctx context.Context) { + // A HTTP/2 CONNECT tunnel is a pair of streams, not a hijackable TCP + // socket. Send the response headers before copying so the peer can start + // sending DATA frames, then flush each upstream write promptly. + w.WriteHeader(http.StatusOK) + if err := http.NewResponseController(w).Flush(); err != nil { + return + } + + done := make(chan struct{}, 2) + go func() { _, _ = io.Copy(upstream, r.Body); done <- struct{}{} }() + go func() { + _, _ = io.Copy(flushingWriter{ResponseWriter: w}, upstream) + done <- struct{}{} + }() + select { + case <-ctx.Done(): + case <-done: + } +} + +type flushingWriter struct { + http.ResponseWriter +} + +func (w flushingWriter) Write(p []byte) (int, error) { + n, err := w.ResponseWriter.Write(p) + if err != nil { + return n, err + } + if err := http.NewResponseController(w.ResponseWriter).Flush(); err != nil { + return n, err + } + return n, nil +} + // Activate allows requests for actorName in atespace. There can be only one // active actor per worker. func (s *Server) Activate(atespace, actorName string) error { @@ -266,48 +390,58 @@ func (s *Server) closeIdleUpstreamConnections() { // ServeHTTP validates the actor hostname on every request before proxying it. func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + _, requestCtx, release, ok := s.authorize(r) + if !ok { + s.reject(w) + return + } + defer release() + + // Do not expose the router-only routing header to actor code. Restore Host + // so dataplanes that route dynamically on worker IP still give the actor its + // stable actor DNS name. + actorHost := r.Header.Get(OriginalHostHeader) + if actorHost == "" { + actorHost = r.Host + } + r.Header.Del(OriginalHostHeader) + r.Host = actorHost + + // ReverseProxy changes the URL destination but intentionally retains Host, + // allowing the actor application to observe its stable actor DNS name. + s.proxy.ServeHTTP(w, r.WithContext(requestCtx)) +} + +func (s *Server) authorize(r *http.Request) (resources.ActorRef, context.Context, func(), bool) { actorHost := r.Header.Get(OriginalHostHeader) if actorHost == "" { actorHost = r.Host } host, err := requestHostname(actorHost) if err != nil { - s.reject(w) - return + return resources.ActorRef{}, nil, nil, false } ref, err := resources.ParseActorDNSName(host) if err != nil { - s.reject(w) - return + return resources.ActorRef{}, nil, nil, false } s.mu.Lock() active := s.active if active == nil || active.ref != ref { s.mu.Unlock() - s.reject(w) - return + return resources.ActorRef{}, nil, nil, false } active.wg.Add(1) s.mu.Unlock() - defer active.wg.Done() - requestCtx, cancel := context.WithCancel(r.Context()) stop := context.AfterFunc(active.ctx, cancel) - defer func() { + release := func() { + active.wg.Done() stop() cancel() - }() - - // Do not expose the router-only routing header to actor code. Restore Host - // so dataplanes that route dynamically on worker IP still give the actor its - // stable actor DNS name. - r.Header.Del(OriginalHostHeader) - r.Host = actorHost - - // ReverseProxy changes the URL destination but intentionally retains Host, - // allowing the actor application to observe its stable actor DNS name. - s.proxy.ServeHTTP(w, r.WithContext(requestCtx)) + } + return ref, requestCtx, release, true } func (s *Server) reject(w http.ResponseWriter) { diff --git a/internal/atunnel/ingress_test.go b/internal/atunnel/ingress_test.go index 56deac913..eeeb2f1f3 100644 --- a/internal/atunnel/ingress_test.go +++ b/internal/atunnel/ingress_test.go @@ -30,6 +30,7 @@ import ( "net/url" "os" "path/filepath" + "slices" "testing" "time" ) @@ -152,6 +153,40 @@ func TestServeHTTPHonorsTargetPortHeader(t *testing.T) { } } +func TestServeConnectHTTPValidatesMethodAndAuthority(t *testing.T) { + upstreamURL, err := url.Parse("http://actor.internal:80") + if err != nil { + t.Fatal(err) + } + s := newTestServer(t, upstreamURL) + if err := s.Activate("team-a", "actor-1"); err != nil { + t.Fatal(err) + } + + tests := []struct { + name string + method string + host string + want int + }{ + {name: "rejects non CONNECT", method: http.MethodGet, host: "actor-1.team-a.actors.resources.substrate.ate.dev:9090", want: http.StatusMethodNotAllowed}, + {name: "requires authority port", method: http.MethodConnect, host: "actor-1.team-a.actors.resources.substrate.ate.dev", want: http.StatusBadRequest}, + {name: "rejects invalid authority port", method: http.MethodConnect, host: "actor-1.team-a.actors.resources.substrate.ate.dev:70000", want: http.StatusMisdirectedRequest}, + {name: "rejects inactive actor", method: http.MethodConnect, host: "actor-2.team-a.actors.resources.substrate.ate.dev:9090", want: http.StatusMisdirectedRequest}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(tt.method, "https://worker/", nil) + req.Host = tt.host + rec := httptest.NewRecorder() + s.ServeConnectHTTP(rec, req) + if rec.Code != tt.want { + t.Fatalf("status = %d, want %d", rec.Code, tt.want) + } + }) + } +} + type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { @@ -252,6 +287,9 @@ func TestMutualTLSClientIdentity(t *testing.T) { if err != nil { t.Fatal(err) } + if got, want := s.tlsConfig.NextProtos, []string{"h2", "http/1.1"}; !slices.Equal(got, want) { + t.Fatalf("ALPN protocols = %v, want %v", got, want) + } untrustedCA := newTestCA(t) tests := []struct { From b34e0ccdfd80226da6f541a428d11d34bd6f45e1 Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Mon, 17 Aug 2026 12:49:23 -0500 Subject: [PATCH 2/8] Fix some flakes Signed-off-by: Keith Mattix II --- cmd/ateapi/internal/controlapi/syncer_test.go | 26 ++++++++++++------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index 1db18baa4..6adf22389 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/cmd/ateapi/internal/controlapi/syncer_test.go @@ -968,11 +968,12 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) { type conflictStore struct { store.Interface conflictTriggered atomic.Bool + shouldInject func(worker *ateapipb.Worker) bool onUpdate func(ctx context.Context, worker *ateapipb.Worker) } func (c *conflictStore) UpdateWorker(ctx context.Context, worker *ateapipb.Worker, expectedVersion int64) error { - if c.onUpdate != nil && c.conflictTriggered.CompareAndSwap(false, true) { + if c.shouldInject != nil && c.shouldInject(worker) && c.conflictTriggered.CompareAndSwap(false, true) { c.onUpdate(ctx, worker) } return c.Interface.UpdateWorker(ctx, worker, expectedVersion) @@ -998,7 +999,20 @@ func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) { var cs *conflictStore persistence, fakeK8s, fakeAte, syncer, cleanup := setupSyncerTestWithStore(t, ctx, func(s store.Interface) store.Interface { - cs = &conflictStore{Interface: s} + // Configure the injector before the syncer starts. It only fires for the + // update under test, so the initial worker creation cannot consume it. + cs = &conflictStore{ + Interface: s, + shouldInject: func(w *ateapipb.Worker) bool { + return w.GetSandboxClass() == "microvm" + }, + onUpdate: func(c context.Context, w *ateapipb.Worker) { + if cw, err := s.GetWorker(c, ns, poolName, podName); err == nil { + cw.NodeName = "node2" + _ = s.UpdateWorker(c, cw, cw.Version) + } + }, + } return cs }, pool) defer func() { @@ -1068,14 +1082,6 @@ func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) { t.Fatalf("pool informer cache failed to update: %v", err) } - // Configure conflictStore to inject a concurrent version bump in Redis when the syncer calls UpdateWorker. - cs.onUpdate = func(c context.Context, w *ateapipb.Worker) { - if cw, err := cs.Interface.GetWorker(c, ns, poolName, podName); err == nil { - cw.NodeName = "node2" - _ = cs.Interface.UpdateWorker(c, cw, cw.Version) - } - } - // Touch the pod ONCE in K8s so the syncer reconciles it. The first reconcile's // UpdateWorker hits ErrVersionConflict (injected by conflictStore), which requeues // the key with backoff; the retry re-fetches the latest version from Redis. From 73b8bd74b59b20214c85c412ccb792dd5c5551d7 Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Mon, 17 Aug 2026 14:08:29 -0500 Subject: [PATCH 3/8] ALPN change was overbroad Signed-off-by: Keith Mattix II --- internal/atunnel/ingress.go | 17 +++++++++-------- internal/atunnel/ingress_test.go | 5 ++--- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/internal/atunnel/ingress.go b/internal/atunnel/ingress.go index 51d3788bc..eaedb7886 100644 --- a/internal/atunnel/ingress.go +++ b/internal/atunnel/ingress.go @@ -156,10 +156,6 @@ func NewServer(cfg Config) (*Server, error) { } s.tlsConfig = &tls.Config{ MinVersion: tls.VersionTLS12, - // HTTP/1.1 CONNECT is relayed by hijacking the connection; HTTP/2 - // CONNECT uses its request and response streams instead. Advertising both - // makes the choice an ALPN negotiation rather than a listener setting. - NextProtos: []string{"h2", "http/1.1"}, GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) { return loadCredentialBundle(s.credentialBundlePath) }, @@ -199,20 +195,25 @@ func loadCredentialBundle(path string) (*tls.Certificate, error) { // Serve serves HTTPS on lis until ctx is canceled or the server fails. func (s *Server) Serve(ctx context.Context, lis net.Listener) error { - return s.serve(ctx, lis, s) + return s.serve(ctx, lis, s, s.tlsConfig) } // ServeConnect serves the mTLS CONNECT endpoint. CONNECT is deliberately on a // separate listener so ordinary actor ingress remains a request proxy, while // the router can use this listener for a bidirectional tunnel. func (s *Server) ServeConnect(ctx context.Context, lis net.Listener) error { - return s.serve(ctx, lis, http.HandlerFunc(s.ServeConnectHTTP)) + // Keep ALPN confined to the CONNECT listener. The established ingress + // listener is explicitly HTTP/1.1 in the router's upstream cluster, while + // the CONNECT listener supports either HTTP/1.1 or HTTP/2. + tlsConfig := s.tlsConfig.Clone() + tlsConfig.NextProtos = []string{"h2", "http/1.1"} + return s.serve(ctx, lis, http.HandlerFunc(s.ServeConnectHTTP), tlsConfig) } -func (s *Server) serve(ctx context.Context, lis net.Listener, handler http.Handler) error { +func (s *Server) serve(ctx context.Context, lis net.Listener, handler http.Handler, tlsConfig *tls.Config) error { httpServer := &http.Server{ Handler: handler, - TLSConfig: s.tlsConfig, + TLSConfig: tlsConfig, ReadHeaderTimeout: 10 * time.Second, } done := make(chan struct{}) diff --git a/internal/atunnel/ingress_test.go b/internal/atunnel/ingress_test.go index eeeb2f1f3..b46900eb0 100644 --- a/internal/atunnel/ingress_test.go +++ b/internal/atunnel/ingress_test.go @@ -30,7 +30,6 @@ import ( "net/url" "os" "path/filepath" - "slices" "testing" "time" ) @@ -287,8 +286,8 @@ func TestMutualTLSClientIdentity(t *testing.T) { if err != nil { t.Fatal(err) } - if got, want := s.tlsConfig.NextProtos, []string{"h2", "http/1.1"}; !slices.Equal(got, want) { - t.Fatalf("ALPN protocols = %v, want %v", got, want) + if got := s.tlsConfig.NextProtos; len(got) != 0 { + t.Fatalf("ordinary ingress ALPN protocols = %v, want none", got) } untrustedCA := newTestCA(t) From c3fd5b53de5d577fac33b2d88947456849511609 Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Mon, 17 Aug 2026 15:08:40 -0500 Subject: [PATCH 4/8] Fix CONNECT test flake Signed-off-by: Keith Mattix II --- .../suites/networking/arbitraryport_test.go | 79 ++++++++++++++++--- 1 file changed, 69 insertions(+), 10 deletions(-) diff --git a/internal/e2e/suites/networking/arbitraryport_test.go b/internal/e2e/suites/networking/arbitraryport_test.go index 59a28750d..12c5dac69 100644 --- a/internal/e2e/suites/networking/arbitraryport_test.go +++ b/internal/e2e/suites/networking/arbitraryport_test.go @@ -17,6 +17,7 @@ package networking import ( "bufio" "context" + "fmt" "io" "net/http" "regexp" @@ -24,6 +25,7 @@ import ( "testing" "time" + "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/resources" ) @@ -62,16 +64,7 @@ func TestActorArbitraryPortAccess(t *testing.T) { }) t.Run("arbitrary port reachable", func(t *testing.T) { - conn, err := router.Connect(ctx, actorRef, counterExtraPort) - if err != nil { - t.Fatalf("CONNECT to the actor's extra port: %v", err) - } - defer conn.Close() - - resp, body := sendTunneledRequest(t, conn, resources.ActorDNSName(actorRef)) - if resp.StatusCode != http.StatusOK { - t.Fatalf("tunneled request returned HTTP %d, want 200; body: %s", resp.StatusCode, body) - } + body := waitForTunneledRouteReady(t, ctx, router, actorRef, counterExtraPort) if !strings.Contains(body, "extra port 9090") { t.Fatalf("tunneled response body = %q, want it to mention extra port 9090", body) } @@ -131,6 +124,50 @@ func TestActorArbitraryPortAccess(t *testing.T) { }) } +// waitForTunneledRouteReady retries an entire CONNECT exchange while the +// resumed actor's route propagates through atenet-router. A CONNECT tunnel is +// tied to one upstream connection, so each retry must establish a fresh tunnel +// rather than reusing a response that Envoy has already closed. +func waitForTunneledRouteReady(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, port int) string { + t.Helper() + const timeout = 30 * time.Second + deadline := time.Now().Add(timeout) + var lastErr error + var lastStatus int + var lastBody string + + for { + conn, err := router.Connect(ctx, actorRef, port) + if err == nil { + resp, body, requestErr := requestTunneled(conn, resources.ActorDNSName(actorRef)) + _ = conn.Close() + if requestErr == nil && resp.StatusCode == http.StatusOK { + return body + } + lastErr = requestErr + if resp != nil { + lastStatus = resp.StatusCode + lastBody = body + } + } else { + lastErr = err + } + + if time.Now().After(deadline) { + if lastErr != nil { + t.Fatalf("tunneled request to port %d did not become ready within %s: %v", port, timeout, lastErr) + } + t.Fatalf("tunneled request to port %d returned HTTP %d after %s, want 200; body: %s", port, lastStatus, timeout, lastBody) + } + if lastErr != nil { + t.Logf("tunneled request to port %d failed: %v; retrying...", port, lastErr) + } else { + t.Logf("tunneled request to port %d returned HTTP %d; retrying...", port, lastStatus) + } + time.Sleep(time.Second) + } +} + // sendTunneledRequest issues a plain GET / over conn (an established CONNECT // tunnel) and returns the parsed response with its body already read and the // connection's read side left consumed accordingly. @@ -155,6 +192,28 @@ func sendTunneledRequest(t *testing.T, conn interface { return resp, string(body) } +func requestTunneled(conn interface { + io.ReadWriter + SetDeadline(time.Time) error +}, host string) (*http.Response, string, error) { + if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil { + return nil, "", fmt.Errorf("setting tunneled request deadline: %w", err) + } + if _, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: " + host + "\r\nConnection: close\r\n\r\n")); err != nil { + return nil, "", fmt.Errorf("writing tunneled request: %w", err) + } + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + return nil, "", fmt.Errorf("reading tunneled response: %w", err) + } + body, err := io.ReadAll(resp.Body) + resp.Body.Close() + if err != nil { + return resp, "", fmt.Errorf("reading tunneled response body (HTTP %d): %w", resp.StatusCode, err) + } + return resp, string(body), nil +} + // counterDefaultPortIPPattern matches the "hello from: | ..." response // counter.go's default (port 80) handler returns. var counterDefaultPortIPPattern = regexp.MustCompile(`^hello from: (\S+)`) From 165454a2c1e60a3a5f53d054ac75d06fe10fb9d3 Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Mon, 17 Aug 2026 16:47:23 -0500 Subject: [PATCH 5/8] Remove unused helper Signed-off-by: Keith Mattix II --- .../suites/networking/arbitraryport_test.go | 24 ------------------- 1 file changed, 24 deletions(-) diff --git a/internal/e2e/suites/networking/arbitraryport_test.go b/internal/e2e/suites/networking/arbitraryport_test.go index 12c5dac69..7121637ce 100644 --- a/internal/e2e/suites/networking/arbitraryport_test.go +++ b/internal/e2e/suites/networking/arbitraryport_test.go @@ -168,30 +168,6 @@ func waitForTunneledRouteReady(t *testing.T, ctx context.Context, router *e2e.Ro } } -// sendTunneledRequest issues a plain GET / over conn (an established CONNECT -// tunnel) and returns the parsed response with its body already read and the -// connection's read side left consumed accordingly. -func sendTunneledRequest(t *testing.T, conn interface { - io.ReadWriter - SetDeadline(time.Time) error -}, host string) (*http.Response, string) { - t.Helper() - conn.SetDeadline(time.Now().Add(10 * time.Second)) - if _, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: " + host + "\r\nConnection: close\r\n\r\n")); err != nil { - t.Fatalf("writing tunneled request: %v", err) - } - resp, err := http.ReadResponse(bufio.NewReader(conn), nil) - if err != nil { - t.Fatalf("reading tunneled response: %v", err) - } - body, err := io.ReadAll(resp.Body) - resp.Body.Close() - if err != nil { - t.Fatalf("reading tunneled response body (HTTP %d): %v", resp.StatusCode, err) - } - return resp, string(body) -} - func requestTunneled(conn interface { io.ReadWriter SetDeadline(time.Time) error From e6bc515b03f1eeb2495d1332de7b0beb1a9e2954 Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Tue, 18 Aug 2026 10:33:22 -0500 Subject: [PATCH 6/8] Fix context propagation and TCP half-close Signed-off-by: Keith Mattix II --- internal/atunnel/egress.go | 27 ++++++--- internal/atunnel/egress_test.go | 97 +++++++++++++++++++++++++++++++++ internal/atunnel/ingress.go | 23 ++------ 3 files changed, 121 insertions(+), 26 deletions(-) diff --git a/internal/atunnel/egress.go b/internal/atunnel/egress.go index c42899c01..1534e5f60 100644 --- a/internal/atunnel/egress.go +++ b/internal/atunnel/egress.go @@ -280,22 +280,35 @@ func (e *Egress) handle(downstream net.Conn) { } func copyBothWays(a, b net.Conn) { + relayWithHalfClose(context.Background(), a, b, b) +} + +// relayWithHalfClose copies a bidirectional stream. Each completed copy +// half-closes its destination, then waits for the other direction to finish. +// This preserves request/response protocols that rely on TCP FIN to delimit a +// request while still receiving a response. +func relayWithHalfClose(ctx context.Context, upstream io.ReadWriter, clientReader io.Reader, clientWriter io.Writer) { done := make(chan struct{}, 2) go func() { - _, _ = io.Copy(a, b) - closeWrite(a) + _, _ = io.Copy(upstream, clientReader) + closeWrite(upstream) done <- struct{}{} }() go func() { - _, _ = io.Copy(b, a) - closeWrite(b) + _, _ = io.Copy(clientWriter, upstream) + closeWrite(clientWriter) done <- struct{}{} }() - <-done - <-done + for range 2 { + select { + case <-ctx.Done(): + return + case <-done: + } + } } -func closeWrite(conn net.Conn) { +func closeWrite(conn any) { if conn, ok := conn.(interface{ CloseWrite() error }); ok { _ = conn.CloseWrite() } diff --git a/internal/atunnel/egress_test.go b/internal/atunnel/egress_test.go index 18440ec5f..196e4b9d0 100644 --- a/internal/atunnel/egress_test.go +++ b/internal/atunnel/egress_test.go @@ -15,6 +15,7 @@ package atunnel import ( + "bytes" "context" "crypto/tls" "errors" @@ -31,6 +32,102 @@ import ( "google.golang.org/grpc/status" ) +func TestRelayWithHalfClose(t *testing.T) { + clientReader, clientInput := io.Pipe() + defer clientReader.Close() + upstream := &closeWriteReadWriter{reader: strings.NewReader("response"), closed: make(chan struct{})} + clientWriter := &closeWriteBuffer{closed: make(chan struct{})} + + done := make(chan struct{}) + go func() { + relayWithHalfClose(context.Background(), upstream, clientReader, clientWriter) + close(done) + }() + + select { + case <-clientWriter.closed: + // The upstream EOF is propagated as a half-close to the client, but the + // relay must still wait for the client to finish its request stream. + case <-time.After(time.Second): + t.Fatal("upstream EOF did not half-close the client stream") + } + select { + case <-done: + t.Fatal("relay returned before the client stream finished") + default: + } + + if _, err := io.WriteString(clientInput, "request"); err != nil { + t.Fatal(err) + } + if err := clientInput.Close(); err != nil { + t.Fatal(err) + } + select { + case <-upstream.closed: + case <-time.After(time.Second): + t.Fatal("client EOF did not half-close the upstream stream") + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("relay did not finish after both streams closed") + } + if got := upstream.String(); got != "request" { + t.Errorf("upstream received %q, want request", got) + } + if got := clientWriter.String(); got != "response" { + t.Errorf("client received %q, want response", got) + } +} + +type closeWriteReadWriter struct { + reader io.Reader + buffer bytes.Buffer + closed chan struct{} +} + +func (c *closeWriteReadWriter) Read(p []byte) (int, error) { + return c.reader.Read(p) +} + +func (c *closeWriteReadWriter) Write(p []byte) (int, error) { + return c.buffer.Write(p) +} + +func (c *closeWriteReadWriter) String() string { + return c.buffer.String() +} + +func (c *closeWriteReadWriter) CloseWrite() error { + if c.closed == nil { + c.closed = make(chan struct{}) + } + select { + case <-c.closed: + default: + close(c.closed) + } + return nil +} + +type closeWriteBuffer struct { + bytes.Buffer + closed chan struct{} +} + +func (c *closeWriteBuffer) CloseWrite() error { + if c.closed == nil { + c.closed = make(chan struct{}) + } + select { + case <-c.closed: + default: + close(c.closed) + } + return nil +} + func TestEgressActivationFailsClosed(t *testing.T) { egress, err := NewEgress(func(net.Conn) (string, error) { return "", nil }) if err != nil { diff --git a/internal/atunnel/ingress.go b/internal/atunnel/ingress.go index eaedb7886..1ec0b8021 100644 --- a/internal/atunnel/ingress.go +++ b/internal/atunnel/ingress.go @@ -21,7 +21,6 @@ import ( "crypto/x509" "errors" "fmt" - "io" "log/slog" "net" "net/http" @@ -256,7 +255,8 @@ func (s *Server) ServeConnectHTTP(w http.ResponseWriter, r *http.Request) { return } - upstream, err := net.DialTimeout("tcp", net.JoinHostPort(s.upstream.Hostname(), port), 5*time.Second) + dialer := &net.Dialer{Timeout: 5 * time.Second} + upstream, err := dialer.DialContext(ctx, "tcp", net.JoinHostPort(s.upstream.Hostname(), port)) if err != nil { slog.WarnContext(r.Context(), "atunnel CONNECT upstream failed", slog.Any("actor", ref), slog.Any("err", err)) http.Error(w, "bad gateway", http.StatusBadGateway) @@ -289,13 +289,7 @@ func (s *Server) serveH1Connect(w http.ResponseWriter, upstream net.Conn, ctx co return } - done := make(chan struct{}, 2) - go func() { _, _ = io.Copy(upstream, rw); done <- struct{}{} }() - go func() { _, _ = io.Copy(client, upstream); done <- struct{}{} }() - select { - case <-ctx.Done(): - case <-done: - } + relayWithHalfClose(ctx, upstream, rw, client) } func (s *Server) serveH2Connect(w http.ResponseWriter, r *http.Request, upstream net.Conn, ctx context.Context) { @@ -307,16 +301,7 @@ func (s *Server) serveH2Connect(w http.ResponseWriter, r *http.Request, upstream return } - done := make(chan struct{}, 2) - go func() { _, _ = io.Copy(upstream, r.Body); done <- struct{}{} }() - go func() { - _, _ = io.Copy(flushingWriter{ResponseWriter: w}, upstream) - done <- struct{}{} - }() - select { - case <-ctx.Done(): - case <-done: - } + relayWithHalfClose(ctx, upstream, r.Body, flushingWriter{ResponseWriter: w}) } type flushingWriter struct { From 4ce7ac1d5e4ab77d166706a41fc2be01e801e80b Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Tue, 18 Aug 2026 11:15:03 -0500 Subject: [PATCH 7/8] Revert ingress changes Signed-off-by: Keith Mattix II --- internal/atunnel/egress.go | 27 +++------ internal/atunnel/egress_test.go | 97 -------------------------------- internal/atunnel/ingress.go | 34 ++++++++++- internal/atunnel/ingress_test.go | 95 +++++++++++++++++++++++++++++++ 4 files changed, 134 insertions(+), 119 deletions(-) diff --git a/internal/atunnel/egress.go b/internal/atunnel/egress.go index 1534e5f60..c42899c01 100644 --- a/internal/atunnel/egress.go +++ b/internal/atunnel/egress.go @@ -280,35 +280,22 @@ func (e *Egress) handle(downstream net.Conn) { } func copyBothWays(a, b net.Conn) { - relayWithHalfClose(context.Background(), a, b, b) -} - -// relayWithHalfClose copies a bidirectional stream. Each completed copy -// half-closes its destination, then waits for the other direction to finish. -// This preserves request/response protocols that rely on TCP FIN to delimit a -// request while still receiving a response. -func relayWithHalfClose(ctx context.Context, upstream io.ReadWriter, clientReader io.Reader, clientWriter io.Writer) { done := make(chan struct{}, 2) go func() { - _, _ = io.Copy(upstream, clientReader) - closeWrite(upstream) + _, _ = io.Copy(a, b) + closeWrite(a) done <- struct{}{} }() go func() { - _, _ = io.Copy(clientWriter, upstream) - closeWrite(clientWriter) + _, _ = io.Copy(b, a) + closeWrite(b) done <- struct{}{} }() - for range 2 { - select { - case <-ctx.Done(): - return - case <-done: - } - } + <-done + <-done } -func closeWrite(conn any) { +func closeWrite(conn net.Conn) { if conn, ok := conn.(interface{ CloseWrite() error }); ok { _ = conn.CloseWrite() } diff --git a/internal/atunnel/egress_test.go b/internal/atunnel/egress_test.go index 196e4b9d0..18440ec5f 100644 --- a/internal/atunnel/egress_test.go +++ b/internal/atunnel/egress_test.go @@ -15,7 +15,6 @@ package atunnel import ( - "bytes" "context" "crypto/tls" "errors" @@ -32,102 +31,6 @@ import ( "google.golang.org/grpc/status" ) -func TestRelayWithHalfClose(t *testing.T) { - clientReader, clientInput := io.Pipe() - defer clientReader.Close() - upstream := &closeWriteReadWriter{reader: strings.NewReader("response"), closed: make(chan struct{})} - clientWriter := &closeWriteBuffer{closed: make(chan struct{})} - - done := make(chan struct{}) - go func() { - relayWithHalfClose(context.Background(), upstream, clientReader, clientWriter) - close(done) - }() - - select { - case <-clientWriter.closed: - // The upstream EOF is propagated as a half-close to the client, but the - // relay must still wait for the client to finish its request stream. - case <-time.After(time.Second): - t.Fatal("upstream EOF did not half-close the client stream") - } - select { - case <-done: - t.Fatal("relay returned before the client stream finished") - default: - } - - if _, err := io.WriteString(clientInput, "request"); err != nil { - t.Fatal(err) - } - if err := clientInput.Close(); err != nil { - t.Fatal(err) - } - select { - case <-upstream.closed: - case <-time.After(time.Second): - t.Fatal("client EOF did not half-close the upstream stream") - } - select { - case <-done: - case <-time.After(time.Second): - t.Fatal("relay did not finish after both streams closed") - } - if got := upstream.String(); got != "request" { - t.Errorf("upstream received %q, want request", got) - } - if got := clientWriter.String(); got != "response" { - t.Errorf("client received %q, want response", got) - } -} - -type closeWriteReadWriter struct { - reader io.Reader - buffer bytes.Buffer - closed chan struct{} -} - -func (c *closeWriteReadWriter) Read(p []byte) (int, error) { - return c.reader.Read(p) -} - -func (c *closeWriteReadWriter) Write(p []byte) (int, error) { - return c.buffer.Write(p) -} - -func (c *closeWriteReadWriter) String() string { - return c.buffer.String() -} - -func (c *closeWriteReadWriter) CloseWrite() error { - if c.closed == nil { - c.closed = make(chan struct{}) - } - select { - case <-c.closed: - default: - close(c.closed) - } - return nil -} - -type closeWriteBuffer struct { - bytes.Buffer - closed chan struct{} -} - -func (c *closeWriteBuffer) CloseWrite() error { - if c.closed == nil { - c.closed = make(chan struct{}) - } - select { - case <-c.closed: - default: - close(c.closed) - } - return nil -} - func TestEgressActivationFailsClosed(t *testing.T) { egress, err := NewEgress(func(net.Conn) (string, error) { return "", nil }) if err != nil { diff --git a/internal/atunnel/ingress.go b/internal/atunnel/ingress.go index 1ec0b8021..9cf824eb1 100644 --- a/internal/atunnel/ingress.go +++ b/internal/atunnel/ingress.go @@ -21,6 +21,7 @@ import ( "crypto/x509" "errors" "fmt" + "io" "log/slog" "net" "net/http" @@ -289,7 +290,7 @@ func (s *Server) serveH1Connect(w http.ResponseWriter, upstream net.Conn, ctx co return } - relayWithHalfClose(ctx, upstream, rw, client) + relayIngressWithHalfClose(ctx, upstream, rw, client, client) } func (s *Server) serveH2Connect(w http.ResponseWriter, r *http.Request, upstream net.Conn, ctx context.Context) { @@ -301,7 +302,36 @@ func (s *Server) serveH2Connect(w http.ResponseWriter, r *http.Request, upstream return } - relayWithHalfClose(ctx, upstream, r.Body, flushingWriter{ResponseWriter: w}) + relayIngressWithHalfClose(ctx, upstream, r.Body, flushingWriter{ResponseWriter: w}, r.Body) +} + +// relayIngressWithHalfClose copies a CONNECT stream. When the client request +// stream ends, it half-closes the actor connection and continues forwarding +// actor output until that stream ends too. +func relayIngressWithHalfClose(ctx context.Context, upstream net.Conn, clientReader io.Reader, clientWriter io.Writer, clientCloser io.Closer) { + stop := context.AfterFunc(ctx, func() { + _ = upstream.Close() + _ = clientCloser.Close() + }) + defer stop() + + done := make(chan struct{}, 2) + go func() { + _, _ = io.Copy(upstream, clientReader) + closeWrite(upstream) + done <- struct{}{} + }() + go func() { + _, _ = io.Copy(clientWriter, upstream) + done <- struct{}{} + }() + for range 2 { + select { + case <-ctx.Done(): + return + case <-done: + } + } } type flushingWriter struct { diff --git a/internal/atunnel/ingress_test.go b/internal/atunnel/ingress_test.go index b46900eb0..a121a39fc 100644 --- a/internal/atunnel/ingress_test.go +++ b/internal/atunnel/ingress_test.go @@ -15,6 +15,7 @@ package atunnel import ( + "bytes" "context" "crypto/ecdsa" "crypto/elliptic" @@ -23,6 +24,7 @@ import ( "crypto/x509" "crypto/x509/pkix" "encoding/pem" + "io" "math/big" "net" "net/http" @@ -34,6 +36,99 @@ import ( "time" ) +func TestRelayIngressWithHalfClose(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + + accepted := make(chan net.Conn, 1) + go func() { + conn, err := listener.Accept() + if err == nil { + accepted <- conn + } + }() + actor, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer actor.Close() + upstream := <-accepted + defer upstream.Close() + + clientReader, clientInput := io.Pipe() + defer clientReader.Close() + var clientOutput bytes.Buffer + done := make(chan struct{}) + go func() { + relayIngressWithHalfClose(context.Background(), upstream, clientReader, &clientOutput, clientReader) + close(done) + }() + + if _, err := io.WriteString(clientInput, "request"); err != nil { + t.Fatal(err) + } + if err := clientInput.Close(); err != nil { + t.Fatal(err) + } + if err := actor.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + request, err := io.ReadAll(actor) + if err != nil { + t.Fatal(err) + } + if got := string(request); got != "request" { + t.Fatalf("actor received %q, want request", got) + } + + if _, err := io.WriteString(actor, "response"); err != nil { + t.Fatal(err) + } + if err := actor.Close(); err != nil { + t.Fatal(err) + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("relay did not finish after the actor response ended") + } + if got := clientOutput.String(); got != "response" { + t.Errorf("client received %q, want response", got) + } +} + +func TestRelayIngressCancellationClosesBothSides(t *testing.T) { + upstream, actor := net.Pipe() + defer actor.Close() + clientReader, clientInput := io.Pipe() + defer clientInput.Close() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + relayIngressWithHalfClose(ctx, upstream, clientReader, io.Discard, clientReader) + close(done) + }() + + cancel() + if err := actor.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, err := actor.Read(make([]byte, 1)); err == nil { + t.Fatal("actor connection remained open after relay cancellation") + } + if _, err := io.WriteString(clientInput, "request"); err == nil { + t.Fatal("client stream remained open after relay cancellation") + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("relay did not return after cancellation") + } +} + func TestServeHTTP(t *testing.T) { upstreamHost := make(chan string, 4) upstreamURL, err := url.Parse("http://actor.internal:80") From 50535a7b8299b3e2665551de7be91c186cb959a8 Mon Sep 17 00:00:00 2001 From: Keith Mattix II Date: Tue, 18 Aug 2026 22:40:29 -0500 Subject: [PATCH 8/8] Bump agentgateway Signed-off-by: Keith Mattix II --- .../ate-install/components/agentgateway/kustomization.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/manifests/ate-install/components/agentgateway/kustomization.yaml b/manifests/ate-install/components/agentgateway/kustomization.yaml index d1cfddc96..9c9150653 100644 --- a/manifests/ate-install/components/agentgateway/kustomization.yaml +++ b/manifests/ate-install/components/agentgateway/kustomization.yaml @@ -55,7 +55,7 @@ patches: path: /spec/template/spec/containers/1 value: name: agentgateway - image: cr.agentgateway.dev/agentgateway:v1.4.1 + image: cr.agentgateway.dev/agentgateway:v0.0.0-alpha.a6c0e366 args: - -f - /etc/agentgateway/config.yaml