diff --git a/builder.go b/builder.go index 71a490c..1f70e77 100644 --- a/builder.go +++ b/builder.go @@ -1,6 +1,7 @@ package main import ( + "bytes" "context" "embed" "errors" @@ -150,10 +151,8 @@ func (s *Builder) Sync(ctx context.Context, _ *builderv0.SyncRequest) (*builderv return s.Base.Builder.SyncError(err) } } - if s.FastAPI.Settings.ConnectEndpoint { - if err := s.writeConnectFace(); err != nil { - return s.Base.Builder.SyncError(err) - } + if err := s.syncConnectFace(); err != nil { + return s.Base.Builder.SyncError(err) } return s.Base.Builder.SyncResponse() } @@ -170,6 +169,35 @@ const connectFacePath = "code/src/rpc/connect_gen.py" //go:embed templates/connect/connect_gen.py var connectFace []byte +// connectFaceMarker is the line every Connect face the agent wrote carries, and +// what lets Sync tell its own file from one the service wrote at that path. +const connectFaceMarker = "This code is generated by the python-fastapi agent (``connect-endpoint: true``)" + +// syncConnectFace writes the Connect face when the service sets +// connect-endpoint and removes the one the agent wrote when it does not: the +// file is the setting's projection, and a face left behind is drift (the +// process serves it only with CODEFLY_CONNECT_PORT, which follows the setting). +func (s *Builder) syncConnectFace() error { + if s.FastAPI.Settings.ConnectEndpoint { + return s.writeConnectFace() + } + target := s.Local(connectFacePath) + content, err := os.ReadFile(target) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return s.Wool.Wrapf(err, "cannot read %s", connectFacePath) + } + if !bytes.Contains(content, []byte(connectFaceMarker)) { + return nil + } + if err := os.Remove(target); err != nil { + return s.Wool.Wrapf(err, "cannot remove the Connect face") + } + return nil +} + // writeConnectFace writes the Connect face, replacing whatever is there: the // file is the agent's, and a hand edit is drift the next sync removes. func (s *Builder) writeConnectFace() error { @@ -473,9 +501,9 @@ type Parameters struct { // binds (the app defaults to it when CODEFLY_GRPC_PORT is unset). GRPCEnabled bool GRPCPort int - // ConnectEnabled adds the Connect face's containerPort and Service port. - // ConnectPort is the fixed in-cluster port it binds (the app defaults to it - // when CODEFLY_CONNECT_PORT is unset). + // ConnectEnabled adds the Connect face's containerPort, Service port and + // CODEFLY_CONNECT_PORT, the variable the app serves the face on. + // ConnectPort is the fixed in-cluster port it binds. ConnectEnabled bool ConnectPort int // AutomountServiceAccountToken projects the ServiceAccount token into the diff --git a/connect_test.go b/connect_test.go index 452c957..3eb8e0c 100644 --- a/connect_test.go +++ b/connect_test.go @@ -2,10 +2,18 @@ package main import ( "context" + "fmt" + "io" + "net" + "net/http" "os" "os/exec" "path/filepath" + "strings" + "sync" + "syscall" "testing" + "time" agenttesting "github.com/codefly-dev/core/agents/testing" agentv0 "github.com/codefly-dev/core/generated/go/codefly/services/agent/v0" @@ -70,11 +78,15 @@ func TestDeploymentRendersConnectPort(t *testing.T) { deployment, err := os.ReadFile(filepath.Join(enabled, "base", "deployment.yaml")) require.NoError(t, err) require.Contains(t, string(deployment), "containerPort: 8081") + require.Contains(t, string(deployment), "- name: CODEFLY_CONNECT_PORT\n value: \"8081\"") disabled := agenttesting.AssertKustomizeTemplates(t, deploymentFS, deploymentTestParameters(t, Parameters{GRPCEnabled: true, GRPCPort: 9090})) svc, err = os.ReadFile(filepath.Join(disabled, "base", "service.yaml")) require.NoError(t, err) require.NotContains(t, string(svc), "connect-port") + deployment, err = os.ReadFile(filepath.Join(disabled, "base", "deployment.yaml")) + require.NoError(t, err) + require.NotContains(t, string(deployment), "CODEFLY_CONNECT_PORT") } // TestConnectFaceServesEveryProtocol runs the face's own suite @@ -130,3 +142,194 @@ func TestBuilderLoadReadsTheAgentsSettings(t *testing.T) { require.True(t, loaded.FastAPI.Settings.ConnectEndpoint) require.Equal(t, filepath.Join(created.Location, "code"), loaded.Service.SourceLocation) } + +// TestSyncConnectFaceFollowsTheSetting proves Sync removes the face it wrote +// once connect-endpoint is off (the scaffold's main starts the face whenever +// the module is there) and leaves a file at that path the agent did not write. +func TestSyncConnectFaceFollowsTheSetting(t *testing.T) { + builder := NewBuilder(NewService()) + builder.Location = t.TempDir() + builder.FastAPI.Settings.GRPCServer.Enabled = true + target := filepath.Join(builder.Location, connectFacePath) + + builder.FastAPI.Settings.ConnectEndpoint = true + require.NoError(t, builder.syncConnectFace()) + require.FileExists(t, target) + + builder.FastAPI.Settings.ConnectEndpoint = false + require.NoError(t, builder.syncConnectFace()) + require.NoFileExists(t, target) + require.NoError(t, builder.syncConnectFace(), "nothing to remove is not an error") + + require.NoError(t, os.WriteFile(target, []byte("# the service's own\n"), 0o644)) + require.NoError(t, builder.syncConnectFace()) + require.FileExists(t, target) +} + +// TestGeneratedMainServesConnect boots a scaffolded gRPC service's own +// src/main.py under uvicorn, as the runtime does. With connect-endpoint on +// (CODEFLY_CONNECT_PORT set, the face synced), the face answers a Connect call +// and stops with the app. With it off (the variable unset), a face left on +// disk binds nothing, not even its 8081 default. On but never synced, the app +// refuses to start and says to sync. +func TestGeneratedMainServesConnect(t *testing.T) { + if testing.Short() { + t.Skip("boots a scaffolded service with uv") + } + if _, err := exec.LookPath("uv"); err != nil { + t.Skip("uv is not installed: the scaffolded service cannot run") + } + builder, root := createServiceForTest(t, map[string]*agentv0.Answer{ + HotReload: confirmAnswer(false), + PublicEndpoint: confirmAnswer(false), + GRPCServer: confirmAnswer(true), + }) + code := filepath.Join(root, "code") + // The stubs Sync generates with Buf, generated the way the scaffold's own + // tests/rpc/conftest.py does when run directly: same plugins, no network + // plugin registry. + generated := filepath.Join(code, "src", "rpc", "_generated") + require.NoError(t, os.MkdirAll(generated, 0o755)) + protoc := exec.Command("uv", "run", "python", "-m", "grpc_tools.protoc", + "-I../proto", "--python_out=src/rpc/_generated", "--grpc_python_out=src/rpc/_generated", "../proto/api.proto") + protoc.Dir = code + out, err := protoc.CombinedOutput() + require.NoError(t, err, "%s", out) + + builder.FastAPI.Settings.ConnectEndpoint = true + require.NoError(t, builder.syncConnectFace()) + app := bootScaffold(t, code, true) + request, err := http.NewRequest(http.MethodPost, + fmt.Sprintf("http://127.0.0.1:%d/api.EchoService/Echo", app.connect), + strings.NewReader(`{"message":"over connect","addedLater":1}`)) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/json") + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + body, err := io.ReadAll(response.Body) + _ = response.Body.Close() + require.NoError(t, err) + require.Equal(t, http.StatusOK, response.StatusCode, "%s", body) + require.JSONEq(t, `{"message":"over connect"}`, string(body)) + app.stop(t) + require.False(t, listens(app.connect), "the face outlived the app") + + // Off, with the face still on disk (no sync since): the variable is the switch. + defaultFree := !listens(8081) + app = bootScaffold(t, code, false) + require.True(t, listens(app.grpc), "the gRPC server is up") + if defaultFree { + require.False(t, listens(8081), "a service without connect-endpoint bound the face's default port") + } + app.stop(t) + + // On, never synced: the app says what is missing instead of serving no face. + require.NoError(t, os.Remove(filepath.Join(root, connectFacePath))) + refused := startScaffold(t, code, true) + select { + case <-refused.exited: + case <-time.After(3 * time.Minute): + t.Fatalf("the app served without its Connect face:\n%s", refused.output) + } + require.Contains(t, refused.output.String(), "src/rpc/connect_gen.py is missing: run `codefly sync service`") +} + +type scaffoldApp struct { + cmd *exec.Cmd + exited chan error + output *lockedOutput + rest, grpc, connect uint16 +} + +// startScaffold launches the scaffold's app under uvicorn with the variables +// the runtime hands it: CODEFLY_CONNECT_PORT only when connect is on. +func startScaffold(t *testing.T, code string, connect bool) *scaffoldApp { + t.Helper() + app := &scaffoldApp{output: &lockedOutput{}, exited: make(chan error, 1), + rest: freePort(t), grpc: freePort(t), connect: freePort(t)} + app.cmd = exec.Command("uv", "run", "uvicorn", "src.main:app", "--host", "127.0.0.1", "--port", fmt.Sprint(app.rest)) + app.cmd.Dir = code + environment := []string{fmt.Sprintf("CODEFLY_GRPC_PORT=%d", app.grpc)} + for _, variable := range os.Environ() { + if !strings.HasPrefix(variable, "CODEFLY_CONNECT_PORT=") { + environment = append(environment, variable) + } + } + if connect { + environment = append(environment, fmt.Sprintf("CODEFLY_CONNECT_PORT=%d", app.connect)) + } + app.cmd.Env = environment + app.cmd.Stdout, app.cmd.Stderr = app.output, app.output + require.NoError(t, app.cmd.Start()) + go func() { app.exited <- app.cmd.Wait() }() + t.Cleanup(func() { _ = app.cmd.Process.Kill() }) + return app +} + +// bootScaffold starts the scaffold's app and returns once it answers on its +// REST port: uvicorn listens only after every startup handler, gRPC and +// Connect included, has run. +func bootScaffold(t *testing.T, code string, connect bool) *scaffoldApp { + t.Helper() + app := startScaffold(t, code, connect) + deadline := time.Now().Add(3 * time.Minute) // a cold uv sync + for !answers(app.rest) { + select { + case err := <-app.exited: + t.Fatalf("the app exited before serving: %v\n%s", err, app.output) + case <-time.After(200 * time.Millisecond): + } + require.True(t, time.Now().Before(deadline), "the app never served:\n%s", app.output) + } + return app +} + +// stop asks uvicorn to shut down, which runs the app's shutdown handlers. +func (a *scaffoldApp) stop(t *testing.T) { + t.Helper() + require.NoError(t, a.cmd.Process.Signal(syscall.SIGTERM)) + select { + case <-a.exited: + case <-time.After(30 * time.Second): + t.Fatalf("the app did not stop:\n%s", a.output) + } + require.NotContains(t, a.output.String(), "Traceback", a.output.String()) +} + +// answers reports whether the scaffold's /version route answers. +func answers(port uint16) bool { + client := http.Client{Timeout: 500 * time.Millisecond} + resp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d/version", port)) + if err != nil { + return false + } + _ = resp.Body.Close() + return resp.StatusCode == http.StatusOK +} + +// lockedOutput is the app's combined output, read while the app writes it. +type lockedOutput struct { + mu sync.Mutex + buf strings.Builder +} + +func (o *lockedOutput) Write(p []byte) (int, error) { + o.mu.Lock() + defer o.mu.Unlock() + return o.buf.Write(p) +} + +func (o *lockedOutput) String() string { + o.mu.Lock() + defer o.mu.Unlock() + return o.buf.String() +} + +func listens(port uint16) bool { + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Second) + if err != nil { + return false + } + _ = conn.Close() + return true +} diff --git a/main.go b/main.go index d0640b6..0464638 100644 --- a/main.go +++ b/main.go @@ -86,9 +86,12 @@ type Settings struct { // speaks Connect and gRPC-Web (HTTP/1.1) and gRPC (HTTP/2) on one port: // the face a browser reaches through an L7 hop. Every call is transcoded // onto the grpc.aio server, so it needs grpc-server and adds no API of its - // own. The agent writes the face (code/src/rpc/connect_gen.py) at sync and - // hands the process its port as CODEFLY_CONNECT_PORT; the app's lifespan - // starts it beside the gRPC server. Off by default. + // own. The agent writes the face (code/src/rpc/connect_gen.py) at sync, + // removes it when the setting is off, and hands the process its port as + // CODEFLY_CONNECT_PORT, locally and in the deployment, exactly when it is + // on: the scaffold's src/main.py starts the face on that variable once the + // gRPC server listens. Start refuses a run where nothing then listens on the + // connect port. Off by default. ConnectEndpoint bool `yaml:"connect-endpoint"` // RuntimeImage overrides the default codefly-built runtime image. diff --git a/runtime.go b/runtime.go index 6709780..ddc7078 100644 --- a/runtime.go +++ b/runtime.go @@ -153,6 +153,10 @@ type Runtime struct { // connectPort is the mapped port for the Connect face, passed to the // Python process as CODEFLY_CONNECT_PORT. Zero when Connect is disabled. connectPort uint16 + // connectAddress is the address Init resolved for the Connect face, which + // Start requires to answer once the HTTP endpoint does. Empty when Connect + // is disabled. + connectAddress string cacheLocation string @@ -506,6 +510,7 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim return s.Base.Runtime.InitError(connectErr) } s.connectPort = uint16(connectNet.Port) + s.connectAddress = connectNet.Address s.Infof("connect will run on %s", connectNet.Address) } @@ -667,7 +672,8 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt } if s.connectPort != 0 { // The same for the Connect face (src/rpc/connect_gen.py), which the - // lifespan starts over the gRPC server; its fixed fallback is 8081. + // app starts over the gRPC server on this variable alone: unset, it + // serves no face. proc.WithEnvironmentVariables(ctx, resources.Env("CODEFLY_CONNECT_PORT", s.connectPort)) } @@ -741,11 +747,7 @@ func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error if s.serveAddress == "" { return nil } - target := s.serveAddress - if !strings.Contains(target, "://") { - target = "http://" + target - } - target = strings.TrimRight(target, "/") + "/" + target := httpTarget(s.serveAddress) timeout := s.serveTimeout if timeout <= 0 { timeout = defaultServeTimeout @@ -762,7 +764,7 @@ func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error err := answers(ctx, target) if err == nil { s.Infof("fastapi app serving on %s after %s", s.serveAddress, time.Since(started).Round(100*time.Millisecond)) - return nil + return s.awaitConnect(ctx, answers) } last = err if !s.runnerAlive(ctx, handle) { @@ -778,6 +780,47 @@ func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error } } +// connectGrace is how long the Connect face may lag the HTTP endpoint. The +// scaffold's main starts it in the startup uvicorn finishes before it listens, +// so it normally answers at once; the grace covers an app that starts it from +// its own lifespan a moment later. +const connectGrace = 5 * time.Second + +// awaitConnect requires the Connect face to answer once the HTTP endpoint +// does. connect-endpoint advertises the port to every consumer, so a process +// that never serves it (a main.py scaffolded before the generated main started +// the face) is refused by name rather than left STARTED with a port that +// refuses connections. The face answers any HTTP request, a GET with 405. +func (s *Runtime) awaitConnect(ctx context.Context, answers func(context.Context, string) error) error { + if s.connectAddress == "" { + return nil + } + target := httpTarget(s.connectAddress) + deadline := time.Now().Add(connectGrace) + for { + err := answers(ctx, target) + if err == nil { + return nil + } + if time.Now().After(deadline) || ctx.Err() != nil { + return fmt.Errorf("%s is on but nothing serves the Connect face at %s: the app must start it once its gRPC server listens, as the scaffold's src/main.py does (connect_gen.start(grpc_port) from src/rpc/connect_gen.py): %v", ConnectEndpointSetting, s.connectAddress, err) + } + select { + case <-ctx.Done(): + case <-time.After(servePollInterval): + } + } +} + +// httpTarget is address as the URL of its root, whether or not it carries a +// scheme. +func httpTarget(address string) string { + if !strings.Contains(address, "://") { + address = "http://" + address + } + return strings.TrimRight(address, "/") + "/" +} + // serveClient is what httpAnswers probes with: one bounded attempt, and a // redirect counts as an answer since the endpoint is serving. var serveClient = &http.Client{ diff --git a/serve_readiness_test.go b/serve_readiness_test.go index adc3af5..98ca41e 100644 --- a/serve_readiness_test.go +++ b/serve_readiness_test.go @@ -125,3 +125,41 @@ func TestStartFailsWhenTheEndpointNeverAnswers(t *testing.T) { require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) require.Len(t, env.started(), 2) } + +// TestStartRefusesAConnectFaceNobodyServes: with connect-endpoint on, an app +// whose HTTP endpoint answers but whose main never starts the Connect face (a +// src/main.py scaffolded before the generated main started it) is a failed +// Start naming the setting and the fix, not a STARTED behind a refused port. +func TestStartRefusesAConnectFaceNobodyServes(t *testing.T) { + server := httptest.NewServer(http.NotFoundHandler()) + t.Cleanup(server.Close) + runtime, env := startTestRuntime(t) + runtime.serveAddress = server.URL + runtime.connectAddress = closedAddress(t) + + resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState()) + require.Contains(t, resp.GetStatus().GetMessage(), "connect-endpoint is on but nothing serves the Connect face") + require.Contains(t, resp.GetStatus().GetMessage(), "connect_gen.start") + require.True(t, env.started()[0].isStopped(), "the app that does not serve its face must be stopped") + require.Nil(t, currentRunner(runtime)) +} + +// TestStartWaitsForTheConnectFace: a face that answers, whatever the status +// (it answers a GET with 405), is a STARTED service. +func TestStartWaitsForTheConnectFace(t *testing.T) { + server := httptest.NewServer(http.NotFoundHandler()) + t.Cleanup(server.Close) + face := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusMethodNotAllowed) + })) + t.Cleanup(face.Close) + runtime, _ := startTestRuntime(t) + runtime.serveAddress = server.URL + runtime.connectAddress = strings.TrimPrefix(face.URL, "http://") + + resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) +} diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py index 73f8d19..1b80f3b 100644 --- a/templates/connect/connect_gen.py +++ b/templates/connect/connect_gen.py @@ -18,12 +18,20 @@ end to end, so they are served on the gRPC protocol only. Request bodies are bounded by ``max_message_bytes`` (gRPC's 4 MiB unless the service -raises it) before they are read, and every read is bounded in time. - -The service starts it from its lifespan, beside its gRPC server, on the port the agent -hands the process as ``CODEFLY_CONNECT_PORT`` (``DEFAULT_PORT`` where the mapping is -fixed). It needs ``grpcio``, ``protobuf`` and ``h11`` (uvicorn's own HTTP/1.1 parser) in -the service's dependencies. +raises it) before they are read, and every read is bounded in time: an HTTP/1.1 read by +``read_timeout`` (``idle_timeout`` between requests), and a relayed gRPC connection is +closed once it has had no open call for ``idle_timeout``, as Go's HTTP/2 server bounds an +idle connection. A call in flight is never cut by the face, however long its handler +takes to answer: its deadline and the gRPC server's own policy bound it. +A JSON request field the service's proto does not know is ignored, as connect-go's codec +ignores it, so a client built from a newer proto calls a Python service and a Go one alike. + +The agent hands the process ``CODEFLY_CONNECT_PORT`` exactly when the service sets +``connect-endpoint``, locally and in its deployment. The agent's generated +``src/main.py`` then calls ``start`` once its gRPC server listens and stops the face +before that server; a service with its own main calls ``serve`` itself. It needs +``grpcio``, ``protobuf`` and ``h11`` (uvicorn's own HTTP/1.1 parser) in the service's +dependencies. """ from __future__ import annotations @@ -32,6 +40,7 @@ import base64 import json import logging +import os import re import struct import zlib @@ -42,15 +51,27 @@ import h11 from google.protobuf import any_pb2, descriptor_pool, json_format, message_factory -__all__ = ["ConnectFace", "serve"] +__all__ = ["ConnectFace", "serve", "start"] PORT_ENV = "CODEFLY_CONNECT_PORT" +# The port a deployment hands the face (CODEFLY_CONNECT_PORT); ``start`` reads only the variable. DEFAULT_PORT = 8081 DEFAULT_MAX_MESSAGE_BYTES = 4 * 1024 * 1024 DEFAULT_READ_TIMEOUT_S = 30.0 DEFAULT_IDLE_TIMEOUT_S = 120.0 _PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" +# An HTTP/2 frame header: 24-bit length, type, flags, stream id (RFC 9113 section 4.1). +_FRAME = struct.Struct(">BHBBI") +_DATA, _HEADERS, _RST_STREAM, _SETTINGS, _GOAWAY = 0x0, 0x1, 0x3, 0x4, 0x7 +_END_STREAM_FLAG, _ACK_FLAG = 0x1, 0x1 +_MAX_CONCURRENT_STREAMS = 0x3 +# Calls one relayed connection may hold open when the gRPC server advertises no +# SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none); past it the face closes the +# connection, so what it tracks per connection is bounded. +DEFAULT_MAX_OPEN_CALLS = 1000 +# The most of a SETTINGS or GOAWAY payload the relay reads; the rest is skipped. +_CONTROL_PAYLOAD = 256 _ENVELOPE = struct.Struct(">BI") _COMPRESSED, _END_STREAM, _TRAILERS = 0x01, 0x02, 0x80 _STATUS_DETAILS = "grpc-status-details-bin" @@ -185,6 +206,75 @@ async def accepted(reader, writer): return face +async def start(grpc_port: int, **options) -> ConnectFace | None: + """Start the face the agent's generated ``src/main.py`` starts: on + ``CODEFLY_CONNECT_PORT`` over the gRPC server on ``grpc_port``, with ``serve``'s + ``options``. Without that variable the service does not set ``connect-endpoint``: + return None, starting nothing, and say why.""" + port = os.environ.get(PORT_ENV) + if not port: + _log.warning( + "connect face: %s is unset, so connect-endpoint is off: not serving", PORT_ENV + ) + return None + return await serve(int(port), grpc_port, **options) + + +class _ProtocolError(Exception): + """A client frame no HTTP/2 client may send: the relay closes the connection.""" + + +class _Frames: + """The HTTP/2 frames in one direction of a relayed connection, after the preface: + ``on_frame(type, flags, stream, payload)`` for each. The payload is read for SETTINGS + and GOAWAY only (at most ``_CONTROL_PAYLOAD`` bytes of it) and skipped otherwise.""" + + def __init__(self, on_frame): + self._on_frame = on_frame + self._header = bytearray() + self._frame = None # (type, flags, stream) while its payload is read + self._payload = bytearray() + self._keep = 0 # payload bytes still to read + self._skip = 0 # payload bytes still to skip + + def feed(self, data: bytes) -> None: + at = 0 + while at < len(data): + if self._keep: + step = min(self._keep, len(data) - at) + self._payload += data[at : at + step] + self._keep -= step + at += step + if not self._keep: + self._emit() + continue + if self._skip: + step = min(self._skip, len(data) - at) + self._skip -= step + at += step + continue + step = min(_FRAME.size - len(self._header), len(data) - at) + self._header += data[at : at + step] + at += step + if len(self._header) == _FRAME.size: + high, low, kind, flags, stream = _FRAME.unpack(self._header) + self._header.clear() + length = high << 16 | low + self._frame = (kind, flags, stream & 0x7FFFFFFF) + if kind in (_SETTINGS, _GOAWAY): + self._keep = min(length, _CONTROL_PAYLOAD) + self._skip = length - self._keep + else: + self._skip = length + if not self._keep: + self._emit() + + def _emit(self): + payload = bytes(self._payload) + self._payload.clear() + self._on_frame(*self._frame, payload) + + class _Connection: def __init__(self, channel, grpc_address, limit, read_timeout, idle_timeout): self.channel, self.grpc_address, self.limit = channel, grpc_address, limit @@ -204,23 +294,88 @@ async def serve(self, reader, writer): await self._http1(reader, writer, head) async def _relay(self, reader, writer, head): - """gRPC over HTTP/2: the bytes go to the gRPC server as they came.""" + """gRPC over HTTP/2: the bytes go to the gRPC server as they came, until either + side closes, or the connection has had no open call for ``idle_timeout``. + + The frame headers say which calls are open: a client HEADERS on a new stream (odd, + above every stream before it) opens one, and the server's END_STREAM, either + side's RST_STREAM, or the server's GOAWAY above the stream ends it. A client that + opens an even stream, or more calls than the server allows at once + (SETTINGS_MAX_CONCURRENT_STREAMS, ``DEFAULT_MAX_OPEN_CALLS`` when it advertises + none), is closed: no HTTP/2 client does either.""" upstream_reader, upstream_writer = await asyncio.open_connection(*self.grpc_address) upstream_writer.write(head) - - async def pump(source, sink): + loop = asyncio.get_running_loop() + last = loop.time() + calls: set[int] = set() + highest = 0 + max_calls = DEFAULT_MAX_OPEN_CALLS + + def from_client(kind, flags, stream, payload): + nonlocal highest + if kind == _HEADERS: + if stream % 2 == 0: + raise _ProtocolError(f"HEADERS on stream {stream}, which a server opens") + if stream <= highest: + # Trailers on a call, or a stream the server refuses as closed: + # never a new call. + return + highest = stream + if len(calls) >= max_calls: + raise _ProtocolError(f"more than {max_calls} calls open at once") + calls.add(stream) + elif kind == _RST_STREAM: + calls.discard(stream) + + def from_server(kind, flags, stream, payload): + nonlocal max_calls + if kind == _RST_STREAM or (kind in (_DATA, _HEADERS) and flags & _END_STREAM_FLAG): + calls.discard(stream) + elif kind == _SETTINGS and not flags & _ACK_FLAG: + for at in range(0, len(payload) - 5, 6): + key, value = struct.unpack_from(">HI", payload, at) + if key == _MAX_CONCURRENT_STREAMS: + max_calls = value + elif kind == _GOAWAY and len(payload) >= 4: + # Streams above the last one the server processes are never answered. + last_stream = struct.unpack_from(">I", payload)[0] & 0x7FFFFFFF + calls.difference_update({call for call in calls if call > last_stream}) + + async def pump(source, sink, frames): + nonlocal last try: while chunk := await source.read(65536): + last = loop.time() + frames.feed(chunk) sink.write(chunk) await sink.drain() - if sink.can_write_eof(): - sink.write_eof() except (ConnectionError, OSError): pass - + except _ProtocolError as refusal: + _log.info("connect face: closing a gRPC connection: %s", refusal) + + async def idle(): + # The clock runs from the last byte either way, and only ends the connection + # while no call is open: a handler computing an answer sends nothing. + while True: + remaining = last + self.idle_timeout - loop.time() + if remaining <= 0 and not calls: + return + await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout) + + tasks = ( + asyncio.ensure_future(pump(reader, upstream_writer, _Frames(from_client))), + asyncio.ensure_future(pump(upstream_reader, writer, _Frames(from_server))), + asyncio.ensure_future(idle()), + ) try: - await asyncio.gather(pump(reader, upstream_writer), pump(upstream_reader, writer)) + # Either side closing ends the relay: an HTTP/2 connection is not half-closed, + # and a call still recorded open cannot be answered once its server has gone. + await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) upstream_writer.close() async def _read(self, reader, timeout): @@ -326,7 +481,10 @@ def request(self, cls, data: bytes) -> bytes: if self.name == "proto": return data try: - return json_format.Parse(data or b"{}", cls()).SerializeToString() + # Unknown fields are dropped, as connect-go's JSON codec does: a client built + # from a newer contract still reaches a service on the older one. + message = json_format.Parse(data or b"{}", cls(), ignore_unknown_fields=True) + return message.SerializeToString() except (json_format.ParseError, UnicodeDecodeError, ValueError) as exc: raise CallError(grpc.StatusCode.INVALID_ARGUMENT, f"unmarshal JSON: {exc}") from exc diff --git a/templates/deployment/kustomize/base/deployment.yaml.tmpl b/templates/deployment/kustomize/base/deployment.yaml.tmpl index faa9d0f..57fca03 100644 --- a/templates/deployment/kustomize/base/deployment.yaml.tmpl +++ b/templates/deployment/kustomize/base/deployment.yaml.tmpl @@ -74,6 +74,12 @@ spec: # writable scratch mount so libraries that use ~/.cache still work. - name: HOME value: /tmp +{{- if .Deployment.Parameters.ConnectEnabled }} + # connect-endpoint is on: the app serves the Connect face on this + # port, and only when the variable is set. + - name: CODEFLY_CONNECT_PORT + value: "{{ .Deployment.Parameters.ConnectPort }}" +{{- end }} {{- if .Restricted }} {{- range $environmentVariable, $reference := .SecretReferences }} - name: {{ $environmentVariable }} diff --git a/templates/factory/code/src/main.py.tmpl b/templates/factory/code/src/main.py.tmpl index 8019333..ce9f825 100644 --- a/templates/factory/code/src/main.py.tmpl +++ b/templates/factory/code/src/main.py.tmpl @@ -59,11 +59,16 @@ for plugin in plugins: # The port is provided by codefly at runtime; it falls back to the standard # gRPC port in container/Kubernetes where the mapping is fixed. _grpc_server = None +# The Connect face (Connect, gRPC-Web and gRPC on one listener over the gRPC +# server): codefly hands the process CODEFLY_CONNECT_PORT exactly when the +# service sets connect-endpoint, and `codefly sync` writes the face itself, +# src/rpc/connect_gen.py. +_connect_face = None @app.on_event("startup") async def _start_grpc(): - global _grpc_server + global _grpc_server, _connect_face # Imported inside the handler, not at module load: the generated stubs # are produced by `codefly sync` and importing this module must not depend # on them (openapi.py imports src.main to emit the schema before sync). @@ -71,10 +76,23 @@ async def _start_grpc(): port = int(os.environ.get("CODEFLY_GRPC_PORT", "9090")) _grpc_server = await _grpc_serve(port) + if os.environ.get("CODEFLY_CONNECT_PORT"): + try: + import src.rpc.connect_gen as connect_gen + except ModuleNotFoundError as missing: + if missing.name != "src.rpc.connect_gen": + raise + raise RuntimeError( + "connect-endpoint is on but src/rpc/connect_gen.py is missing: " + "run `codefly sync service`" + ) from missing + _connect_face = await connect_gen.start(port) @app.on_event("shutdown") async def _stop_grpc(): + if _connect_face is not None: + await _connect_face.stop() if _grpc_server is not None: await _grpc_server.stop(grace=5) {{- end }} diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py index 3459d33..ad54917 100644 --- a/testdata/connect/test_connect_gen.py +++ b/testdata/connect/test_connect_gen.py @@ -284,3 +284,250 @@ async def test_one_connection_serves_several_calls(face): assert chunk, seen seen += chunk writer.close() + + +async def test_connect_json_ignores_unknown_fields(face): + # connect-go's JSON codec discards unknown fields: a client built from a newer + # contract reaches a Python service exactly as it reaches a Go one. + async with httpx.AsyncClient() as client: + r = await client.post( + _url(face, "Say"), + json={"text": "hi", "count": 1, "addedLater": {"x": 1}}, + headers={"content-type": "application/json", **STAMP}, + ) + assert r.status_code == 200, r.text + assert r.json() == {"text": "hi|maya", "count": 1} + + +async def test_a_silent_grpc_connection_is_closed_after_the_idle_bound(): + # An upstream that accepts and never answers: only the face's own bound ends it. + upstream = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0) + listener = await connect_gen.serve( + 0, + upstream.sockets[0].getsockname()[1], + host="127.0.0.1", + read_timeout=0.5, + idle_timeout=0.5, + ) + try: + reader, writer = await asyncio.open_connection("127.0.0.1", listener.port) + writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") + await writer.drain() + assert await asyncio.wait_for(reader.read(), 3) == b"" + writer.close() + finally: + await listener.stop() + upstream.close() + + +async def test_a_grpc_connection_carrying_traffic_outlives_the_idle_bound(): + # Bytes in either direction keep the relay open: a server stream whose client sends + # nothing is not idle. + async def drip(reader, writer): + for _ in range(6): + writer.write(b".") + await writer.drain() + await asyncio.sleep(0.2) + writer.close() + + upstream = await asyncio.start_server(drip, "127.0.0.1", 0) + listener = await connect_gen.serve( + 0, upstream.sockets[0].getsockname()[1], host="127.0.0.1", idle_timeout=0.5 + ) + try: + reader, writer = await asyncio.open_connection("127.0.0.1", listener.port) + writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") + await writer.drain() + assert await asyncio.wait_for(reader.read(), 3) == b"......" + writer.close() + finally: + await listener.stop() + upstream.close() + + +async def test_start_serves_only_on_the_agents_port(monkeypatch): + server = grpc.aio.server() + server.add_generic_rpc_handlers((_handler(),)) + grpc_port = server.add_insecure_port("127.0.0.1:0") + await server.start() + try: + # No CODEFLY_CONNECT_PORT: the service does not set connect-endpoint, whatever + # file is on disk, so nothing binds. + monkeypatch.delenv(connect_gen.PORT_ENV, raising=False) + assert await connect_gen.start(grpc_port, host="127.0.0.1") is None + monkeypatch.setenv(connect_gen.PORT_ENV, "0") + started = await connect_gen.start(grpc_port, host="127.0.0.1") + assert started is not None + try: + async with httpx.AsyncClient() as client: + r = await client.post( + f"http://127.0.0.1:{started.port}/echo.v1.Echo/Say", + json={"text": "s"}, + headers={"content-type": "application/json", "x-caller": "zoe"}, + ) + assert r.json() == {"text": "s|zoe"} + finally: + await started.stop() + finally: + await server.stop(None) + + +@pytest.fixture +async def short_idle(): + """A face whose idle bound (0.5 s) is shorter than the slow call (2 s).""" + server = grpc.aio.server(interceptors=(Proof(),)) + server.add_generic_rpc_handlers((_handler(),)) + grpc_port = server.add_insecure_port("127.0.0.1:0") + await server.start() + listener = await connect_gen.serve( + 0, grpc_port, host="127.0.0.1", read_timeout=0.5, idle_timeout=0.5 + ) + try: + yield listener + finally: + await listener.stop() + await server.stop(None) + + +async def test_a_grpc_call_in_flight_outlives_the_idle_bound(short_idle): + # The handler sends nothing for 2 s: an open call is never idle. + async with grpc.aio.insecure_channel(f"127.0.0.1:{short_idle.port}") as channel: + say_call = channel.unary_unary( + "/echo.v1.Echo/Say", + request_serializer=Msg.SerializeToString, + response_deserializer=Msg.FromString, + ) + answer = await say_call(Msg(text="slow"), metadata=tuple(STAMP.items()), timeout=10) + assert answer.text == "slow|maya" + + +async def test_a_grpc_connection_with_no_open_call_is_closed(short_idle): + async with grpc.aio.insecure_channel(f"127.0.0.1:{short_idle.port}") as channel: + say_call = channel.unary_unary( + "/echo.v1.Echo/Say", + request_serializer=Msg.SerializeToString, + response_deserializer=Msg.FromString, + ) + assert (await say_call(Msg(text="q"), metadata=tuple(STAMP.items()))).text == "q|maya" + # The call ended: the relayed connection is idle, and closed past the bound. + for _ in range(30): + if not short_idle._connections: + break + await asyncio.sleep(0.1) + assert not short_idle._connections + + +PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + + +def h2_frame(kind, flags, stream, payload=b""): + size = len(payload) + return struct.pack(">BHBBI", size >> 16, size & 0xFFFF, kind, flags, stream) + payload + + +SETTINGS = h2_frame(0x4, 0, 0) + + +def opens(stream): + """A HEADERS frame opening ``stream`` (END_HEADERS; one indexed field).""" + return h2_frame(0x1, 0x4, stream, b"\x82") + + +@pytest.fixture +async def silent_upstream(): + """A gRPC address that accepts, reads everything and never answers, so only the face's + own rules end a connection relayed to it, and ``face(**options)`` serving over it. + ``state["close_after"]`` makes it hang up after that many seconds instead.""" + held, faces = [], [] + state = {"close_after": None} + + async def accept(reader, writer): + held.append(writer) + if state["close_after"] is not None: + await asyncio.sleep(state["close_after"]) + writer.close() + return + while await reader.read(65536): + pass + + upstream = await asyncio.start_server(accept, "127.0.0.1", 0) + port = upstream.sockets[0].getsockname()[1] + + async def face(**options): + listener = await connect_gen.serve(0, port, host="127.0.0.1", **options) + faces.append(listener) + return listener + + try: + yield face, state + finally: + for listener in faces: + await listener.stop() + for writer in held: + writer.close() + upstream.close() + + +async def closed_within(listener, payload, seconds): + reader, writer = await asyncio.open_connection("127.0.0.1", listener.port) + writer.write(payload) + await writer.drain() + try: + async with asyncio.timeout(seconds): + while await reader.read(65536): + pass + return True + except TimeoutError: + return False + finally: + writer.close() + + +async def test_a_client_opening_an_even_stream_is_closed(silent_upstream): + # A client never opens an even stream; grpcio ignores one silently, so a face that + # recorded it as a call would hold the connection, and its memory, forever. + face, _ = silent_upstream + payload = PREFACE + SETTINGS + b"".join(opens(2 * n) for n in range(1, 1000)) + assert await closed_within(await face(), payload, 2) + + +async def test_open_calls_on_one_connection_are_bounded(silent_upstream): + face, _ = silent_upstream + listener = await face() + calls = connect_gen.DEFAULT_MAX_OPEN_CALLS + within = PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls)) + assert not await closed_within(listener, within, 1) + beyond = PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls + 1)) + assert await closed_within(listener, beyond, 2) + + +async def test_a_stream_id_below_the_highest_opens_no_call(silent_upstream): + # 1 and 99 are calls; 3 after 99 is not a new one. Once 1 and 99 are reset no call is + # open, and the connection is closed past the idle bound. + face, _ = silent_upstream + reset = lambda stream: h2_frame(0x3, 0, stream, b"\x00\x00\x00\x08") # noqa: E731 + payload = PREFACE + SETTINGS + opens(1) + opens(99) + opens(3) + reset(1) + reset(99) + assert await closed_within(await face(idle_timeout=0.5), payload, 2) + + +async def test_the_relay_ends_when_the_grpc_server_hangs_up(silent_upstream): + # A call is recorded open and the client is silent: the server closing its side + # still ends the connection, rather than leaving the client's open. + face, state = silent_upstream + state["close_after"] = 0.3 + listener = await face() + reader, writer = await asyncio.open_connection("127.0.0.1", listener.port) + writer.write(PREFACE + SETTINGS + opens(1)) + await writer.drain() + try: + await asyncio.sleep(0.1) + assert listener._connections, "the face took the connection" + await asyncio.sleep(0.3) # the server hangs up + for _ in range(20): + if not listener._connections: + break + await asyncio.sleep(0.1) + assert not listener._connections, "the face still holds the client's connection" + assert await asyncio.wait_for(reader.read(), 1) == b"" + finally: + writer.close()