feat: add TCP transport (--tcp) to worker

Bump vgi-rpc-go to v0.10.0 (adds Server.RunTcp) and expose the raw
Arrow-IPC TCP transport through the worker:

- Worker.RunTcp(host, port, idleTimeout) serves via Server.RunTcp and
  prints the TCP:<host>:<port> launcher discovery line with the actual
  bound port (so ephemeral port==0 binds are discoverable).
- New transportTCP server class (concurrent connections, no cleanup
  hook — same as unix).
- Example worker accepts --tcp [HOST:]PORT (host defaults to
  127.0.0.1), mutually exclusive with --http/--unix. Registered in
  filterKnownFlags as a value-consuming flag so the space-separated
  form parses.

Raw TCP framing carries no auth/encryption (loopback/trusted networks
only); HTTP remains the transport for untrusted networks.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rustyconover

cmd/vgi-example-worker/main.go

@@ -9,6 +9,7 @@ import (
 	"log"
 	"math"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -26,7 +27,8 @@ import (
 func main() {
 	httpMode := flag.Bool("http", false, "Run as HTTP server instead of stdio")
 	unixPath := flag.String("unix", "", "Bind to this AF_UNIX socket path (launcher transport); mutually exclusive with --http")
-	idleTimeout := flag.Float64("idle-timeout", 300, "Self-shutdown after N seconds idle when serving --unix (0 = never)")
+	tcpAddr := flag.String("tcp", "", "Bind a raw TCP socket ([HOST:]PORT, host defaults to 127.0.0.1, port 0 auto-selects); mutually exclusive with --http/--unix")
+	idleTimeout := flag.Float64("idle-timeout", 300, "Self-shutdown after N seconds idle when serving --unix/--tcp (0 = never)")
 	// --describe / --no-describe: accepted for launcher compatibility (the VGI
 	// extension passes it through). Description pages aren't served over the
 	// socket/stdio transports, so it is currently a no-op here.
@@ -38,6 +40,7 @@ func main() {
 	// rather than failing to start. Flags named here consume a value token.
 	flag.CommandLine.Parse(filterKnownFlags(os.Args[1:], map[string]bool{
 		"unix":         true,
+		"tcp":          true,
 		"idle-timeout": true,
 		"log-level":    true,
 		"log-format":   true,
@@ -52,8 +55,14 @@ func main() {
 	// (no-op otherwise); the harness kills pooled/long-lived workers with SIGTERM.
 	covflush.Start()
 
-	if *unixPath != "" && *httpMode {
-		log.Fatal("--unix and --http are mutually exclusive")
+	nTransports := 0
+	for _, on := range []bool{*unixPath != "", *tcpAddr != "", *httpMode} {
+		if on {
+			nTransports++
+		}
+	}
+	if nTransports > 1 {
+		log.Fatal("--unix, --tcp, and --http are mutually exclusive")
 	}
 
 	// Pick the FunctionStorage backend from VGI_WORKER_SHARED_STORAGE.
@@ -936,6 +945,14 @@ func main() {
 		if err := w.RunUnix(*unixPath, time.Duration(*idleTimeout*float64(time.Second))); err != nil {
 			log.Fatal(err)
 		}
+	case *tcpAddr != "":
+		host, port, err := parseTCPAddr(*tcpAddr)
+		if err != nil {
+			log.Fatal(err)
+		}
+		if err := w.RunTcp(host, port, time.Duration(*idleTimeout*float64(time.Second))); err != nil {
+			log.Fatal(err)
+		}
 	case *httpMode:
 		authFn, jwtCleanup := resolveAuthenticate()
 		if jwtCleanup != nil {
@@ -955,6 +972,24 @@ func main() {
 	}
 }
 
+// parseTCPAddr parses a "[HOST:]PORT" --tcp bind spec. A bare PORT (no colon)
+// binds 127.0.0.1; an empty host (leading ":") also defaults to loopback.
+func parseTCPAddr(spec string) (string, int, error) {
+	host := "127.0.0.1"
+	portStr := spec
+	if i := strings.LastIndex(spec, ":"); i >= 0 {
+		if spec[:i] != "" {
+			host = spec[:i]
+		}
+		portStr = spec[i+1:]
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		return "", 0, fmt.Errorf("--tcp expects [HOST:]PORT, got %q", spec)
+	}
+	return host, port, nil
+}
+
 // filterKnownFlags drops command-line tokens for flags this binary doesn't
 // define, so launcher-injected argv-differentiation flags (e.g. --threaded,
 // --quiet, --debug) don't abort flag parsing. Flags named in valueFlags consume

go.mod

@@ -10,7 +10,7 @@ go 1.25.0
 replace github.com/apache/arrow-go/v18 => github.com/Query-farm/arrow-go/v18 v18.0.0-20260220022719-2d45cbd918a4
 
 require (
-	github.com/Query-farm/vgi-rpc-go v0.9.4
+	github.com/Query-farm/vgi-rpc-go v0.10.0
 	github.com/Query-farm/vgi-rpc-go/vgirpc/jwtauth v0.9.3
 	github.com/apache/arrow-go/v18 v18.5.2
 	github.com/duckdb/duckdb-go/v2 v2.10502.0

go.sum

@@ -4,8 +4,8 @@ github.com/MicahParks/keyfunc/v3 v3.3.5 h1:7ceAJLUAldnoueHDNzF8Bx06oVcQ5CfJnYwNt
 github.com/MicahParks/keyfunc/v3 v3.3.5/go.mod h1:SdCCyMJn/bYqWDvARspC6nCT8Sk74MjuAY22C7dCST8=
 github.com/Query-farm/arrow-go/v18 v18.0.0-20260220022719-2d45cbd918a4 h1:hH37wAlNZG3LEfHxwnlE9aMtFqZndL26cHOEP0nhWJ8=
 github.com/Query-farm/arrow-go/v18 v18.0.0-20260220022719-2d45cbd918a4/go.mod h1:IJTMBTlHe7cDOhRh0ioGuEKBl5iTR6xPfl5BN4AgirU=
-github.com/Query-farm/vgi-rpc-go v0.9.4 h1:Da+0bNrQkTH2rHXfDZ3i52UldUov4xJx+fWe26n/1Lk=
-github.com/Query-farm/vgi-rpc-go v0.9.4/go.mod h1:XbQBjp31eFKIZaBqmg0q9r2Mxp/LTPeAG97w1qy5gmI=
+github.com/Query-farm/vgi-rpc-go v0.10.0 h1:ddteS2eF/F8Y7ZeyOfRpwjYosVv3MTTyad9U57RuaTs=
+github.com/Query-farm/vgi-rpc-go v0.10.0/go.mod h1:XbQBjp31eFKIZaBqmg0q9r2Mxp/LTPeAG97w1qy5gmI=
 github.com/Query-farm/vgi-rpc-go/vgirpc/jwtauth v0.9.3 h1:9bmMKdhik5PFir5FBQO0yw8dnWBBeDfVpMY5dmQQAGk=
 github.com/Query-farm/vgi-rpc-go/vgirpc/jwtauth v0.9.3/go.mod h1:orAtn+oBHDFucdCpnVfvwtUwQSTjWkNrJ+lV3FsP4q4=
 github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=

vgi/worker.go

@@ -859,6 +859,7 @@ const (
 	transportStdio serverTransport = iota // single serial stream — deferred cleanup
 	transportHTTP                         // independent requests — no deferred cleanup (TTL sweep)
 	transportUnix                         // concurrent connections — no cleanup hook (TTL sweep)
+	transportTCP                          // concurrent connections — no cleanup hook (TTL sweep)
 )
 
 func (w *Worker) buildServer(transport serverTransport) *vgirpc.Server {
@@ -903,6 +904,8 @@ func (w *Worker) buildServer(transport serverTransport) *vgirpc.Server {
 		s.SetDispatchHook(&storageCleanupHook{worker: w, isHTTP: true})
 	case transportUnix:
 		// no cleanup hook
+	case transportTCP:
+		// no cleanup hook (concurrent connections, like unix)
 	}
 
 	// Register bind (unary)
@@ -959,6 +962,23 @@ func (w *Worker) RunUnix(path string, idleTimeout time.Duration) error {
 	})
 }
 
+// RunTcp runs the worker serving RPC over a raw TCP socket — the launcher
+// transport's AF_INET sibling. It binds host:port (host "" defaults to
+// 127.0.0.1; port 0 picks a free port), prints the readiness marker
+// "TCP:<host>:<port>" with the actual bound port to stdout once listening
+// (then writes nothing further to stdout), and self-shuts-down after
+// idleTimeout with no active connections (<=0 disables the timeout).
+//
+// Raw TCP framing carries no auth/encryption — bind loopback / a trusted
+// network only; use RunHttp for untrusted networks.
+func (w *Worker) RunTcp(host string, port int, idleTimeout time.Duration) error {
+	s := w.buildServer(transportTCP)
+	return s.RunTcp(host, port, idleTimeout, func(boundHost string, boundPort int) {
+		fmt.Printf("TCP:%s:%d\n", boundHost, boundPort)
+		os.Stdout.Sync()
+	})
+}
+
 // RunHttp runs the worker serving RPC over HTTP. It listens on the given
 // address (e.g. "127.0.0.1:0" for a random port) and prints "PORT:<n>" to
 // stdout so callers can discover the assigned port.