Add opt-in TCP port readiness check for upstream process#130
Open
le0pard wants to merge 1 commit into
Open
Conversation
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds an option to delay starting the proxy until the upstream binds its target port, and introduces more robust upstream lifecycle handling (signals/termination) with accompanying configuration and tests.
Changes:
- Add
WAIT_FOR_TARGET_PORT+ timeout config, plus README documentation. - Run upstream asynchronously and optionally block until the target port opens before starting the proxy.
- Add tests for
waitForTargetPortand extend config default/override tests.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| internal/service.go | Adds upstream result channel, port-waiting logic, and signal-based termination handling. |
| internal/service_test.go | Introduces tests for the new waitForTargetPort behavior. |
| internal/config.go | Adds config fields + env parsing for waiting on the target port. |
| internal/config_test.go | Verifies new config defaults and env overrides. |
| README.md | Documents the new environment variables. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+103
to
+105
| case result := <-resultChan: | ||
| // The upstream process crashed before the port was opened | ||
| return fmt.Errorf("upstream process exited prematurely with code %d: %w", result.exitCode, result.err) |
Comment on lines
+139
to
+145
| // Wait up to 10 seconds for graceful shutdown of upstream | ||
| select { | ||
| case <-resultChan: | ||
| slog.Info("Upstream process terminated gracefully after signal.") | ||
| case <-time.After(10 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 10 seconds of signal.") | ||
| } |
Comment on lines
+119
to
+120
| signalChan := make(chan os.Signal, 1) | ||
| signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM) |
Comment on lines
+118
to
+154
| func (s *Service) awaitTermination(upstream *UpstreamProcess, resultChan <-chan upstreamResult) int { | ||
| signalChan := make(chan os.Signal, 1) | ||
| signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM) | ||
|
|
||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Wrapped process exited on its own.", "exit_code", result.exitCode) | ||
| if result.err != nil { | ||
| slog.Error("Wrapped process failed", "command", s.config.UpstreamCommand, "args", s.config.UpstreamArgs, "error", result.err) | ||
| return 1 | ||
| } | ||
| return result.exitCode | ||
|
|
||
| case sig := <-signalChan: | ||
| slog.Info("Received signal, shutting down.", "signal", sig.String()) | ||
| slog.Info("Relaying signal to upstream process...") | ||
|
|
||
| if err := upstream.Signal(sig); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } | ||
|
|
||
| // Wait up to 10 seconds for graceful shutdown of upstream | ||
| select { | ||
| case <-resultChan: | ||
| slog.Info("Upstream process terminated gracefully after signal.") | ||
| case <-time.After(10 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 10 seconds of signal.") | ||
| } | ||
|
|
||
| // Exit with a conventional code for signals (128 + signal number) | ||
| sysSig, ok := sig.(syscall.Signal) | ||
| if ok { | ||
| return 128 + int(sysSig) | ||
| } | ||
| return 1 | ||
| } | ||
| } |
Comment on lines
+17
to
+20
| listener, err := net.Listen("tcp", "127.0.0.1:0") | ||
| require.NoError(t, err) | ||
| port := listener.Addr().(*net.TCPAddr).Port | ||
| listener.Close() |
Comment on lines
+139
to
+141
| if err := upstream.Signal(sig); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } |
Comment on lines
+56
to
79
| resultChan := make(chan upstreamResult, 1) | ||
|
|
||
| // Start the upstream process | ||
| go func() { | ||
| exitCode, err := upstream.Run() | ||
| resultChan <- upstreamResult{exitCode: exitCode, err: err} | ||
| }() | ||
|
|
||
| if s.config.WaitForTargetPort { | ||
| if err := s.waitForTargetPort(resultChan); err != nil { | ||
| slog.Error("Failed waiting for target port", "error", err) | ||
|
|
||
| // Try to gracefully kill the upstream process if it didn't exit already | ||
| if err := upstream.Signal(syscall.SIGTERM); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } | ||
| return 1 | ||
| } | ||
| slog.Info("Upstream service is bound to port, starting proxy server.") | ||
| } | ||
|
|
||
| if err := server.Start(); err != nil { | ||
| return 1 | ||
| } |
Comment on lines
+136
to
+141
| // Mock catching a global OS signal | ||
| go func() { | ||
| // Give awaitTermination a brief moment to register signal handlers | ||
| time.Sleep(50 * time.Millisecond) | ||
| _ = syscall.Kill(os.Getpid(), syscall.SIGTERM) | ||
| }() |
Comment on lines
+16
to
+26
| // Helper to get a random free port for each test to prevent cross-test contamination | ||
| func getFreePort(t *testing.T) int { | ||
| t.Helper() | ||
| listener, err := net.Listen("tcp", "127.0.0.1:0") | ||
| require.NoError(t, err) | ||
| port := listener.Addr().(*net.TCPAddr).Port | ||
|
|
||
| // Addressed review comment: explicitly assert close is successful | ||
| require.NoError(t, listener.Close()) | ||
| return port | ||
| } |
Comment on lines
+136
to
+142
| case result := <-resultChan: | ||
| slog.Info("Wrapped process exited on its own.", "exit_code", result.exitCode) | ||
| if result.err != nil { | ||
| slog.Error("Wrapped process failed", "command", s.config.UpstreamCommand, "args", s.config.UpstreamArgs, "error", result.err) | ||
| return 1 | ||
| } | ||
| return result.exitCode |
Comment on lines
+63
to
+71
| if err := s.waitForTargetPort(resultChan, net.DialTimeout); err != nil { | ||
| slog.Error("Failed waiting for target port", "error", err) | ||
|
|
||
| // Try to gracefully kill the upstream process if it didn't exit already | ||
| if err := upstream.Signal(syscall.SIGTERM); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } | ||
| return 1 | ||
| } |
Comment on lines
+32
to
+33
| client, _ := net.Pipe() | ||
| return client, nil |
| }, | ||
| } | ||
| resultChan := make(chan upstreamResult, 1) | ||
| signalChan := make(chan os.Signal, 1) // Unused in this test block |
Comment on lines
+111
to
+117
| service := &Service{ | ||
| config: &Config{ | ||
| UpstreamCommand: "sleep", | ||
| UpstreamArgs: []string{"10"}, | ||
| }, | ||
| } | ||
| upstream := NewUpstreamProcess("sleep", "10") |
Comment on lines
+126
to
+131
| case result := <-resultChan: | ||
| // The upstream process crashed before the port was opened | ||
| if result.err != nil { | ||
| return fmt.Errorf("upstream process exited prematurely with code %d: %w", result.exitCode, result.err) | ||
| } | ||
| return fmt.Errorf("upstream process exited prematurely with code %d", result.exitCode) |
Comment on lines
+117
to
+125
| // Check every 100ms | ||
| ticker := time.NewTicker(100 * time.Millisecond) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return fmt.Errorf("timed out after %v waiting for port %d", s.config.WaitForTargetPortTimeout, s.config.TargetPort) | ||
|
|
Comment on lines
+133
to
+139
| case <-ticker.C: | ||
| // Attempt a fast TCP connection | ||
| conn, err := dial("tcp", addr, 100*time.Millisecond) | ||
| if err == nil { | ||
| conn.Close() // Port is open | ||
| return nil | ||
| } |
Comment on lines
95
to
+105
| if err := server.Start(); err != nil { | ||
| return 1 | ||
| } | ||
| defer server.Stop() | ||
|
|
||
| s.setEnvironment() | ||
|
|
||
| exitCode, err := upstream.Run() | ||
| if err != nil { | ||
| slog.Error("Failed to start wrapped process", "command", s.config.UpstreamCommand, "args", s.config.UpstreamArgs, "error", err) | ||
| return 1 | ||
| } | ||
| // Initialize the signal channel here so it can be managed cleanly | ||
| signalChan := make(chan os.Signal, 1) | ||
| signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM) | ||
| defer signal.Stop(signalChan) | ||
|
|
||
| return exitCode | ||
| return s.awaitTermination(upstream, resultChan, signalChan) |
Comment on lines
+110
to
+135
| func TestService_awaitTermination_Signal(t *testing.T) { | ||
| service := &Service{ | ||
| config: &Config{ | ||
| UpstreamCommand: "sleep", | ||
| UpstreamArgs: []string{"10"}, | ||
| }, | ||
| } | ||
| upstream := NewUpstreamProcess("sleep", "10") | ||
| resultChan := make(chan upstreamResult, 1) | ||
| signalChan := make(chan os.Signal, 1) | ||
|
|
||
| // Start the mocked upstream process | ||
| go func() { | ||
| exitCode, err := upstream.Run() | ||
| resultChan <- upstreamResult{exitCode: exitCode, err: err} | ||
| }() | ||
| <-upstream.Started // Ensure it is ready to receive signals | ||
|
|
||
| // Mock injecting an OS signal directly into the injected channel | ||
| // to avoid terminating the test runner or impacting other tests | ||
| signalChan <- syscall.SIGTERM | ||
|
|
||
| exitCode := service.awaitTermination(upstream, resultChan, signalChan) | ||
|
|
||
| // Ensure the signal relay returned the correct exit code | ||
| assert.Equal(t, 128+int(syscall.SIGTERM), exitCode) |
Comment on lines
+91
to
+95
| | `HTTP_READ_TIMEOUT` | The maximum time in seconds that a client can take to send the request headers and body. | 30 | | ||
| | `HTTP_WRITE_TIMEOUT` | The maximum time in seconds during which the client must read the response. | 30 | | ||
| | `H2C_ENABLED` | Set to `1` or `true` to enable h2c (http/2 cleartext) | Disabled | | ||
| | `WAIT_FOR_TARGET_PORT` | If set to `1` or `true`, Thruster will wait for the upstream application to bind to its port before starting the proxy server. | Disabled | | ||
| | `WAIT_FOR_TARGET_PORT_TIMEOUT` | The maximum time in seconds to wait for the upstream port to open. | 60 | |
Comment on lines
+68
to
+87
| if s.config.WaitForTargetPort { | ||
| if err := s.waitForTargetPort(resultChan, net.DialTimeout); err != nil { | ||
| slog.Error("Failed waiting for target port", "error", err) | ||
|
|
||
| // Try to gracefully kill the upstream process if it didn't exit already | ||
| if err := upstream.Signal(syscall.SIGTERM); err != nil { | ||
| slog.Error("Failed to send SIGTERM to upstream process", "error", err) | ||
| } | ||
|
|
||
| // Wait for the upstream to actually exit, escalate to SIGKILL if it doesn't | ||
| select { | ||
| case <-resultChan: | ||
| slog.Info("Upstream process terminated after SIGTERM.") | ||
| case <-time.After(5 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 5 seconds, sending SIGKILL.") | ||
| _ = upstream.Signal(syscall.SIGKILL) | ||
| } | ||
|
|
||
| return 1 | ||
| } |
Comment on lines
+112
to
+149
| ctx, cancel := context.WithTimeout(context.Background(), s.config.WaitForTargetPortTimeout) | ||
| defer cancel() | ||
|
|
||
| addr := fmt.Sprintf("127.0.0.1:%d", s.config.TargetPort) | ||
| slog.Info("Waiting for upstream to bind to port", "address", addr) | ||
|
|
||
| // Attempt a fast TCP connection immediately to prevent | ||
| // unnecessary latency or false timeouts for fast services | ||
| if conn, err := dial("tcp", addr, 100*time.Millisecond); err == nil { | ||
| conn.Close() // Port is open | ||
| return nil | ||
| } | ||
|
|
||
| return exitCode | ||
| // Fallback to checking every 100ms | ||
| ticker := time.NewTicker(100 * time.Millisecond) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return fmt.Errorf("timed out after %v waiting for port %d", s.config.WaitForTargetPortTimeout, s.config.TargetPort) | ||
|
|
||
| case result := <-resultChan: | ||
| // The upstream process crashed before the port was opened | ||
| if result.err != nil { | ||
| return fmt.Errorf("upstream process exited prematurely with code %d: %w", result.exitCode, result.err) | ||
| } | ||
| return fmt.Errorf("upstream process exited prematurely with code %d", result.exitCode) | ||
|
|
||
| case <-ticker.C: | ||
| // Attempt a fast TCP connection | ||
| conn, err := dial("tcp", addr, 100*time.Millisecond) | ||
| if err == nil { | ||
| conn.Close() // Port is open | ||
| return nil | ||
| } | ||
| } | ||
| } |
Comment on lines
+90
to
+99
| // Non-blocking check to ensure the upstream didn't fail immediately on boot. | ||
| time.Sleep(10 * time.Millisecond) | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Error("Upstream process exited prematurely", "command", s.config.UpstreamCommand, "exit_code", result.exitCode, "error", result.err) | ||
| return result.exitCode | ||
| default: | ||
| // Upstream is running (or didn't fail instantly), proceed | ||
| } | ||
| } |
Comment on lines
+176
to
+182
| // Wait up to 10 seconds for graceful shutdown of upstream | ||
| select { | ||
| case <-resultChan: | ||
| slog.Info("Upstream process terminated gracefully after signal.") | ||
| case <-time.After(10 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 10 seconds of signal.") | ||
| } |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
Comments suppressed due to low confidence (1)
internal/service.go:1
- Signals are queued in
signalChan“early”, but duringwaitForTargetPort(...)the service doesn’t select onsignalChan(or a signal-cancelled context). This means a SIGINT/SIGTERM received while waiting for the port can be delayed until after the wait completes (potentially up toWaitForTargetPortTimeout), which is incorrect shutdown behavior. A concrete fix is to create acontext.Contextcancelled by signals (e.g., viasignal.NotifyContext) and pass it intowaitForTargetPortso it can return early when a shutdown signal arrives (and then terminate the upstream promptly).
package internal
Comment on lines
55
to
+128
| server := NewServer(s.config, handler) | ||
| upstream := NewUpstreamProcess(s.config.UpstreamCommand, s.config.UpstreamArgs...) | ||
|
|
||
| s.setEnvironment() | ||
|
|
||
| resultChan := make(chan upstreamResult, 1) | ||
|
|
||
| // Start the upstream process | ||
| go func() { | ||
| exitCode, err := upstream.Run() | ||
| resultChan <- upstreamResult{exitCode: exitCode, err: err} | ||
| }() | ||
|
|
||
| if s.config.WaitForTargetPort { | ||
| dial := func(ctx context.Context, network, address string) (net.Conn, error) { | ||
| var d net.Dialer | ||
| return d.DialContext(ctx, network, address) | ||
| } | ||
|
|
||
| if err := s.waitForTargetPort(resultChan, dial); err != nil { | ||
| slog.Error("Failed waiting for target port", "error", err) | ||
|
|
||
| // Check if the upstream has already exited (waitForTargetPort puts the result back) | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Upstream process is already dead.") | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| default: | ||
| // Upstream is still running but port never opened, kill it | ||
| if err := upstream.Signal(syscall.SIGTERM); err != nil { | ||
| slog.Error("Failed to send SIGTERM to upstream process", "error", err) | ||
| } | ||
|
|
||
| select { | ||
| case <-resultChan: | ||
| slog.Info("Upstream process terminated after SIGTERM.") | ||
| case <-time.After(5 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 5 seconds, sending SIGKILL.") | ||
| if upstream != nil { | ||
| _ = upstream.Signal(syscall.SIGKILL) | ||
| select { | ||
| case <-resultChan: | ||
| case <-time.After(1 * time.Second): | ||
| } | ||
| } | ||
| } | ||
| return 1 | ||
| } | ||
| } | ||
| slog.Info("Upstream service is bound to port, starting proxy server.") | ||
| } else { | ||
| // Non-blocking wait for up to 10ms to catch fast command failures without penalizing success. | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Error("Upstream process exited prematurely", "command", s.config.UpstreamCommand, "exit_code", result.exitCode, "error", result.err) | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| case <-time.After(10 * time.Millisecond): | ||
| // Upstream is running (or didn't fail instantly), proceed | ||
| } | ||
| } | ||
|
|
||
| if err := server.Start(); err != nil { | ||
| return 1 | ||
| } | ||
| defer server.Stop() | ||
|
|
||
| s.setEnvironment() | ||
| return s.awaitTermination(upstream, resultChan, signalChan) | ||
| } |
Comment on lines
+210
to
+220
| case <-time.After(10 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 10 seconds of signal.") | ||
| if upstream != nil { | ||
| slog.Warn("Escalating to SIGKILL.") | ||
| _ = upstream.Signal(syscall.SIGKILL) | ||
| // Wait briefly for it to be reaped | ||
| select { | ||
| case <-resultChan: | ||
| case <-time.After(1 * time.Second): | ||
| } | ||
| } |
Comment on lines
+158
to
+166
| addr := fmt.Sprintf("127.0.0.1:%d", s.config.TargetPort) | ||
| slog.Info("Waiting for upstream to bind to port", "address", addr) | ||
|
|
||
| tryDial := func() bool { | ||
| // Cap the dial attempt to 100ms or whatever time is remaining on the parent ctx | ||
| dialCtx, cancelDial := context.WithTimeout(ctx, 100*time.Millisecond) | ||
| defer cancelDial() | ||
|
|
||
| conn, err := dial(dialCtx, "tcp", addr) |
Comment on lines
+223
to
+230
| relaySig := sig | ||
| // Map SIGTERM to os.Interrupt for cross-platform relay safety | ||
| if sysSig, ok := sig.(syscall.Signal); ok && sysSig == syscall.SIGTERM { | ||
| relaySig = os.Interrupt | ||
| } | ||
| if err := upstream.Signal(relaySig); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } |
Comment on lines
+250
to
+254
| // Exit with a conventional code for signals (128 + signal number) | ||
| if sysSig, ok := sig.(syscall.Signal); ok { | ||
| return 128 + int(sysSig) | ||
| } | ||
| return 1 |
Comment on lines
+94
to
+95
| case <-time.After(5 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 5 seconds, killing it.") |
| } | ||
| slog.Info("Upstream service is bound to port, starting proxy server.") | ||
| } else { | ||
| // Non-blocking wait for up to 10ms to catch fast command failures without penalizing success. |
|
|
||
| tryDial := func() bool { | ||
| // Cap the dial attempt to 100ms or whatever time is remaining on the parent ctx | ||
| dialCtx, cancelDial := context.WithTimeout(ctx, 100*time.Millisecond) |
| } | ||
|
|
||
| // Fallback to checking every 100ms | ||
| ticker := time.NewTicker(100 * time.Millisecond) |
| } | ||
| } | ||
|
|
||
| // Wait up to 10 seconds for graceful shutdown of upstream |
| select { | ||
| case <-resultChan: | ||
| slog.Info("Upstream process terminated gracefully after signal.") | ||
| case <-time.After(10 * time.Second): |
Comment on lines
+74
to
+106
| if err := s.waitForTargetPort(resultChan, dial); err != nil { | ||
| slog.Error("Failed waiting for target port", "error", err) | ||
|
|
||
| // Check if the upstream has already exited (waitForTargetPort puts the result back) | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Upstream process is already dead.") | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| default: | ||
| // Upstream is still running but port never opened, shut it down | ||
| if err := upstream.Signal(os.Interrupt); err != nil { | ||
| slog.Error("Failed to send interrupt to upstream process", "error", err) | ||
| } | ||
|
|
||
| select { | ||
| case <-resultChan: | ||
| slog.Info("Upstream process terminated after interrupt.") | ||
| case <-time.After(5 * time.Second): | ||
| slog.Warn("Upstream process did not terminate within 5 seconds, killing it.") | ||
| if upstream != nil && upstream.cmd != nil && upstream.cmd.Process != nil { | ||
| _ = upstream.cmd.Process.Kill() | ||
| select { | ||
| case <-resultChan: | ||
| case <-time.After(1 * time.Second): | ||
| } | ||
| } | ||
| } | ||
| return 1 | ||
| } | ||
| } |
Comment on lines
+82
to
+97
| if err := s.waitForTargetPort(resultChan, signalChan, s.dial); err != nil { | ||
| slog.Error("Failed waiting for target port", "error", err) | ||
|
|
||
| // Check if the upstream has already exited (waitForTargetPort puts the result back) | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Upstream process is already dead.") | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| default: | ||
| // Upstream is still running but port never opened, shut it down | ||
| if err := upstream.Signal(os.Interrupt); err != nil { | ||
| slog.Error("Failed to send interrupt to upstream process", "error", err) | ||
| } |
| ctx, cancel := context.WithTimeout(context.Background(), s.config.WaitForTargetPortTimeout) | ||
| defer cancel() | ||
|
|
||
| addr := fmt.Sprintf("localhost:%d", s.config.TargetPort) |
| dialCtx, cancelDial := context.WithTimeout(ctx, dialTimeout) | ||
| defer cancelDial() | ||
|
|
||
| conn, err := dial(dialCtx, "tcp", addr) |
Comment on lines
+212
to
+214
| case result := <-resultChan: | ||
| // The upstream process crashed before the port was opened | ||
| resultChan <- result // Put it back so Run() can cleanly exit without waiting for SIGTERM |
| } | ||
| return 1 | ||
| case <-time.After(sigtermEscalationTimeout): | ||
| slog.Warn(fmt.Sprintf("Upstream process did not terminate within %v, killing it.", sigtermEscalationTimeout)) |
Comment on lines
+95
to
+118
| if err := upstream.Signal(os.Interrupt); err != nil { | ||
| slog.Error("Failed to send interrupt to upstream process", "error", err) | ||
| } | ||
|
|
||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Upstream process terminated after interrupt.") | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| case <-time.After(sigtermEscalationTimeout): | ||
| slog.Warn(fmt.Sprintf("Upstream process did not terminate within %v, killing it.", sigtermEscalationTimeout)) | ||
| if upstream != nil && upstream.cmd != nil && upstream.cmd.Process != nil { | ||
| _ = upstream.cmd.Process.Kill() | ||
| select { | ||
| case result := <-resultChan: | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| case <-time.After(sigkillWaitTimeout): | ||
| } | ||
| } | ||
| } |
Comment on lines
+124
to
+134
| // Non-blocking wait to catch fast command failures without penalizing success. | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Error("Upstream process exited prematurely", "command", s.config.UpstreamCommand, "exit_code", result.exitCode, "error", result.err) | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| case <-time.After(fastFailureWait): | ||
| // Upstream is running (or didn't fail instantly), proceed | ||
| } |
Comment on lines
+147
to
+148
| os.Setenv("GO_WANT_HELPER_PROCESS", "1") | ||
| defer os.Unsetenv("GO_WANT_HELPER_PROCESS") |
Comment on lines
+231
to
+232
| case sig := <-signalChan: | ||
| slog.Info("Received signal, shutting down.", "signal", sig.String()) |
Comment on lines
+82
to
+98
| if s.config.WaitForTargetPort { | ||
| if err := s.waitForTargetPort(resultChan, signalChan, s.dial); err != nil { | ||
| slog.Error("Failed waiting for target port", "error", err) | ||
|
|
||
| // Check if the upstream has already exited (waitForTargetPort puts the result back) | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Upstream process is already dead.") | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| default: | ||
| // Upstream is still running but port never opened, shut it down | ||
| return s.terminateUpstream(upstream, resultChan, os.Interrupt, sigtermEscalationTimeout, 1) | ||
| } | ||
| } |
Comment on lines
+201
to
+207
| case result := <-resultChan: | ||
| // The upstream process crashed before the port was opened | ||
| resultChan <- result // Put it back so Run() can cleanly exit without waiting for SIGTERM | ||
| if result.err != nil { | ||
| return fmt.Errorf("upstream process exited prematurely with code %d: %w", result.exitCode, result.err) | ||
| } | ||
| return fmt.Errorf("upstream process exited prematurely with code %d", result.exitCode) |
Comment on lines
+143
to
+146
| slog.Warn("Upstream process did not terminate within timeout, killing it.", "timeout", timeout) | ||
| if upstream.cmd != nil && upstream.cmd.Process != nil { | ||
| _ = upstream.cmd.Process.Kill() | ||
| select { |
Comment on lines
+125
to
+126
| sigtermEscalationTimeout = 50 * time.Millisecond | ||
| sigkillWaitTimeout = 50 * time.Millisecond |
| UpstreamArgs: []string{"-test.run=TestHelperProcess"}, | ||
| TargetPort: 3000, | ||
| WaitForTargetPort: true, | ||
| WaitForTargetPortTimeout: 10 * time.Millisecond, |
Comment on lines
+50
to
+52
| signalChan := make(chan os.Signal, 1) | ||
| signal.Notify(signalChan, os.Interrupt, syscall.SIGTERM) | ||
| defer signal.Stop(signalChan) |
Comment on lines
+245
to
+247
| if sysSig, ok := sig.(syscall.Signal); ok { | ||
| fallback = 128 + int(sysSig) | ||
| } |
|
|
||
| // Block until signal is received | ||
| c := make(chan os.Signal, 1) | ||
| signal.Notify(c, os.Interrupt, syscall.SIGTERM) |
Comment on lines
+255
to
+257
| if sysSig, ok := sig.(syscall.Signal); ok { | ||
| os.Exit(128 + int(sysSig)) | ||
| } |
|
|
||
| // Mock injecting an OS signal directly into the injected channel | ||
| // We relay unmodified, so sending SIGTERM will hit the helper which catches SIGTERM | ||
| signalChan <- syscall.SIGTERM |
| exitCode := service.awaitTermination(upstream, resultChan, signalChan) | ||
|
|
||
| // Ensure the signal relay returned the correct exit code observed from the upstream | ||
| expectedExitCode := 128 + int(syscall.SIGTERM) |
Comment on lines
+75
to
+77
| | Variable Name | Description | Default Value | | ||
| |------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------| | ||
| | `TLS_DOMAIN` | Comma-separated list of domain names to use for TLS provisioning. If not set, TLS will be disabled. | None | |
Comment on lines
16
to
22
| var ( | ||
| dialTimeout = 100 * time.Millisecond | ||
| fastFailureWait = 10 * time.Millisecond | ||
| gracefulShutdownTimeout = 10 * time.Second | ||
| sigtermEscalationTimeout = 5 * time.Second | ||
| sigkillWaitTimeout = 1 * time.Second | ||
| ) |
Comment on lines
+147
to
150
| slog.Info("Sending signal to upstream process...", "signal", sig) | ||
| if err := upstream.Signal(sig); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } |
Comment on lines
45
to
52
| func (p *UpstreamProcess) handleSignals() { | ||
| ch := make(chan os.Signal, 1) | ||
| signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM) | ||
| signal.Notify(ch, terminationSignals...) | ||
|
|
||
| sig := <-ch | ||
| slog.Info("Relaying signal to upstream process", "signal", sig.String()) | ||
| slog.Info("Relaying signal to upstream process", "signal", sig) | ||
| _ = p.Signal(sig) | ||
| } |
Comment on lines
+12
to
+14
| func exitCodeFromSignal(sig os.Signal) int { | ||
| return 1 | ||
| } |
Comment on lines
+15
to
+21
| type serviceTimeouts struct { | ||
| dial time.Duration | ||
| fastFailureWait time.Duration | ||
| gracefulShutdown time.Duration | ||
| sigtermEscalation time.Duration | ||
| sigkillWait time.Duration | ||
| } |
Comment on lines
+148
to
+154
| timeouts: serviceTimeouts{ | ||
| dial: 100 * time.Millisecond, | ||
| fastFailureWait: 10 * time.Millisecond, | ||
| gracefulShutdown: 500 * time.Millisecond, | ||
| sigtermEscalation: 500 * time.Millisecond, | ||
| sigkillWait: 500 * time.Millisecond, | ||
| }, |
Comment on lines
+24
to
+31
| func forwardSignal(p *os.Process, sig os.Signal) error { | ||
| if sig == os.Interrupt { | ||
| // os.Interrupt is not reliably supported via os.Process.Signal on Windows. | ||
| // Escalating to Kill ensures the child process gracefully and actually terminates. | ||
| return p.Kill() | ||
| } | ||
| return p.Signal(sig) | ||
| } |
Comment on lines
+42
to
+45
| if p.cmd == nil || p.cmd.Process == nil { | ||
| return nil | ||
| } | ||
| return forwardSignal(p.cmd.Process, sig) |
Comment on lines
+26
to
+27
| // os.Interrupt is not reliably supported via os.Process.Signal on Windows. | ||
| // Escalating to Kill ensures the child process gracefully and actually terminates. |
Comment on lines
+75
to
+76
| | Variable Name | Description | Default Value | | ||
| |------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------| |
Comment on lines
+120
to
+129
| select { | ||
| case result := <-resultChan: | ||
| slog.Error("Upstream process exited prematurely", "command", s.config.UpstreamCommand, "exit_code", result.exitCode, "error", result.err) | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| case <-time.After(s.timeouts.fastFailureWait): | ||
| // Upstream is running (or didn't fail instantly), proceed | ||
| } |
Comment on lines
+264
to
+283
| func (s *Service) awaitTermination(upstream *UpstreamProcess, resultChan <-chan upstreamResult, signalChan <-chan os.Signal) int { | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Wrapped process exited on its own.", "exit_code", result.exitCode) | ||
| if result.err != nil { | ||
| slog.Error("Wrapped process failed", "command", s.config.UpstreamCommand, "args", s.config.UpstreamArgs, "error", result.err) | ||
| // Return the upstream's exit code if available, fallback to 1 otherwise | ||
| if result.exitCode != 0 { | ||
| return result.exitCode | ||
| } | ||
| return 1 | ||
| } | ||
| return result.exitCode | ||
|
|
||
| case sig := <-signalChan: | ||
| slog.Info("Received signal, shutting down.", "signal", sig) | ||
| fallback := exitCodeFromSignal(sig) | ||
| return s.terminateUpstream(upstream, resultChan, sig, s.timeouts.gracefulShutdown, fallback) | ||
| } | ||
| } |
Comment on lines
+33
to
+34
| // Broadcast that the process has successfully started | ||
| close(p.Started) |
Comment on lines
+49
to
+50
| if exitErr, ok := err.(*exec.ExitError); ok { | ||
| return exitCodeFromProcessState(exitErr.ProcessState), nil |
Comment on lines
12
to
16
| type UpstreamProcess struct { | ||
| Started chan struct{} | ||
| cmd *exec.Cmd | ||
| Started chan struct{} | ||
| cmd *exec.Cmd | ||
| startOnce sync.Once | ||
| } |
Comment on lines
+25
to
+29
| func forwardSignal(p *os.Process, sig os.Signal) error { | ||
| // Attempt the signal gracefully. If unsupported or ignored on Windows, | ||
| // the higher-level timeout logic will safely escalate to os.Kill. | ||
| return p.Signal(sig) | ||
| } |
Comment on lines
+60
to
+68
| <-p.Started // Ensure the process is fully initialized | ||
|
|
||
| err := p.Signal(os.Interrupt) | ||
| assert.NoError(t, err) | ||
| assert.Equal(t, 128+int(syscall.SIGTERM), exitCode) | ||
|
|
||
| <-done | ||
|
|
||
| assert.NoError(t, runErr) | ||
| assert.Equal(t, exitCodeFromSignal(os.Interrupt), exitCode) |
Comment on lines
+9
to
+11
| // Windows does not have SIGTERM natively, so we only listen for Interrupt | ||
| var terminationSignals = []os.Signal{os.Interrupt} | ||
| var defaultTerminationSignal = os.Interrupt |
Comment on lines
+25
to
+29
| func forwardSignal(p *os.Process, sig os.Signal) error { | ||
| // Attempt the signal gracefully. If unsupported or ignored on Windows, | ||
| // the higher-level timeout logic will safely escalate to os.Kill. | ||
| return p.Signal(sig) | ||
| } |
Comment on lines
+171
to
+177
| timer := time.NewTimer(timeout) | ||
| select { | ||
| case <-resultChan: | ||
| stopTimer(timer) | ||
| slog.Info("Upstream process terminated after signal.") | ||
| // Return the wrapper-specified exit code since we initiated this termination path | ||
| return fallbackExitCode |
Comment on lines
29
to
+42
| @@ -31,36 +36,26 @@ func (p *UpstreamProcess) Run() (int, error) { | |||
| return 0, err | |||
| } | |||
|
|
|||
| p.Started <- struct{}{} | |||
| // Broadcast that the process has successfully started | |||
| p.startOnce.Do(func() { | |||
| close(p.started) | |||
| }) | |||
Comment on lines
+28
to
+36
| func forwardSignal(p *os.Process, sig os.Signal) error { | ||
| if sig == os.Interrupt { | ||
| // os.Interrupt is not reliably supported via os.Process.Signal on Windows unless | ||
| // CREATE_NEW_PROCESS_GROUP is used with specific Windows console APIs. | ||
| // Escalating to Kill ensures the child process gracefully and actually terminates. | ||
| return p.Kill() | ||
| } | ||
| return p.Signal(sig) | ||
| } |
| if sig == os.Interrupt { | ||
| // os.Interrupt is not reliably supported via os.Process.Signal on Windows unless | ||
| // CREATE_NEW_PROCESS_GROUP is used with specific Windows console APIs. | ||
| // Escalating to Kill ensures the child process gracefully and actually terminates. |
Comment on lines
34
to
43
| err := p.cmd.Start() | ||
|
|
||
| // Broadcast that the start attempt has concluded (unblocks waiters on success OR failure) | ||
| p.startOnce.Do(func() { | ||
| close(p.started) | ||
| }) | ||
|
|
||
| if err != nil { | ||
| return 0, err | ||
| } |
Comment on lines
+72
to
+77
| select { | ||
| case <-p.Started(): | ||
| // Process is fully initialized | ||
| case <-time.After(2 * time.Second): | ||
| t.Fatal("timed out waiting for upstream to start") | ||
| } |
Comment on lines
+175
to
+185
| func TestService_Run_EscalatesToKill(t *testing.T) { | ||
| service := &Service{ | ||
| config: &Config{ | ||
| UpstreamCommand: os.Args[0], | ||
| UpstreamArgs: []string{"-test.run=TestHelperProcess"}, | ||
| TargetPort: 3000, | ||
| WaitForTargetPort: true, | ||
| WaitForTargetPortTimeout: 150 * time.Millisecond, | ||
| CacheSizeBytes: 1024, | ||
| MaxCacheItemSizeBytes: 1024, | ||
| }, |
Comment on lines
+362
to
+364
| // Ensure the signal relay returned the correct exit code observed from the upstream | ||
| expectedExitCode := exitCodeFromSignal(os.Interrupt) | ||
| assert.Equal(t, expectedExitCode, exitCode) |
Comment on lines
+230
to
233
| // Attempt a fast TCP connection immediately to prevent unnecessary latency on boot | ||
| if tryDial() { | ||
| return nil, nil, nil | ||
| } |
Comment on lines
+9
to
+22
| // Windows does not have SIGTERM natively, and os.Interrupt is unreliable via os.Process.Signal. | ||
| var terminationSignals = []os.Signal{os.Interrupt} | ||
| var defaultTerminationSignal = os.Kill | ||
|
|
||
| func exitCodeFromSignal(sig os.Signal) int { | ||
| if sig == os.Kill { | ||
| return 137 | ||
| } | ||
| if sig == os.Interrupt { | ||
| // Map Interrupt to POSIX convention 130 for cross-platform consistency | ||
| return 130 | ||
| } | ||
| return 1 | ||
| } |
Comment on lines
+168
to
+190
| timer := time.NewTimer(timeout) | ||
| select { | ||
| case result := <-resultChan: | ||
| stopTimer(timer) | ||
| slog.Info("Upstream process terminated after signal.") | ||
| return resolveExitCode(result, fallbackExitCode) | ||
| case <-timer.C: | ||
| slog.Warn("Upstream process did not terminate within timeout, killing it.", "timeout", timeout) | ||
|
|
||
| // Map safely to os.Kill to honor the UpstreamProcess cross-platform encapsulation | ||
| if err := upstream.Signal(os.Kill); err != nil { | ||
| slog.Error("Failed to send KILL signal to upstream process", "error", err) | ||
| } | ||
|
|
||
| exitCode, err := upstream.Run() | ||
| if err != nil { | ||
| slog.Error("Failed to start wrapped process", "command", s.config.UpstreamCommand, "args", s.config.UpstreamArgs, "error", err) | ||
| return 1 | ||
| killTimer := time.NewTimer(s.timeouts.sigkillWait) | ||
| select { | ||
| case result := <-resultChan: | ||
| stopTimer(killTimer) | ||
| return resolveExitCode(result, fallbackExitCode) | ||
| case <-killTimer.C: | ||
| // Ensure we do not orphan the running process, but provide a hard upper bound | ||
| slog.Error("Upstream process still running after os.Kill, waiting for OS to reap it...", "timeout", s.timeouts.finalReapWait) | ||
| finalTimer := time.NewTimer(s.timeouts.finalReapWait) |
Comment on lines
+13
to
+23
| func exitCodeFromSignal(sig os.Signal) int { | ||
| if sig == os.Kill { | ||
| return 137 | ||
| } | ||
| if sig == os.Interrupt { | ||
| // os.Interrupt escalates to Kill via forwardSignal on Windows. | ||
| // Align the fallback exit code to actual OS behavior. | ||
| return 1 | ||
| } | ||
| return 1 | ||
| } |
Comment on lines
+177
to
+188
| if res, ok := waitResult(resultChan, s.timeouts.sigkillWait); ok { | ||
| return resolveExitCode(res, fallbackExitCode) | ||
| } | ||
|
|
||
| exitCode, err := upstream.Run() | ||
| if err != nil { | ||
| slog.Error("Failed to start wrapped process", "command", s.config.UpstreamCommand, "args", s.config.UpstreamArgs, "error", err) | ||
| return 1 | ||
| // Ensure we do not orphan the running process, but provide a hard upper bound | ||
| slog.Error("Upstream process still running after os.Kill, waiting for OS to reap it...", "timeout", s.timeouts.finalReapWait) | ||
| if res, ok := waitResult(resultChan, s.timeouts.finalReapWait); ok { | ||
| return resolveExitCode(res, fallbackExitCode) | ||
| } | ||
|
|
||
| return exitCode | ||
| slog.Error("Upstream process completely unresponsive, exiting wrapper and abandoning child.") | ||
| return fallbackExitCode |
Comment on lines
+222
to
+253
| // Fallback to checking continuously | ||
| ticker := time.NewTicker(s.timeouts.portCheckInterval) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return nil, nil, fmt.Errorf("timed out after %v waiting for port %d", s.config.WaitForTargetPortTimeout, s.config.TargetPort) | ||
|
|
||
| case sig := <-signalChan: | ||
| return nil, sig, fmt.Errorf("received signal %v while waiting for target port", sig) | ||
|
|
||
| case result := <-resultChan: | ||
| if result.err != nil { | ||
| return &result, nil, fmt.Errorf("upstream process exited prematurely with code %d: %w", result.exitCode, result.err) | ||
| } | ||
| return &result, nil, fmt.Errorf("upstream process exited prematurely with code %d", result.exitCode) | ||
|
|
||
| case <-ticker.C: | ||
| for _, addr := range addrs { | ||
| // Cap the dial attempt to prevent unnecessary blocks on individual checks | ||
| dialCtx, cancelDial := context.WithTimeout(ctx, s.timeouts.dialTimeout) | ||
| conn, err := dial(dialCtx, "tcp", addr) | ||
| cancelDial() | ||
|
|
||
| if err == nil { | ||
| conn.Close() // Port is open | ||
| return nil, nil, nil | ||
| } | ||
| } | ||
| } | ||
| } |
Comment on lines
29
to
+42
| @@ -31,36 +36,26 @@ func (p *UpstreamProcess) Run() (int, error) { | |||
| return 0, err | |||
| } | |||
|
|
|||
| p.Started <- struct{}{} | |||
| // Broadcast that the process has successfully started | |||
| p.startOnce.Do(func() { | |||
| close(p.started) | |||
| }) | |||
Comment on lines
34
to
+44
| err := p.cmd.Start() | ||
| if err != nil { | ||
| close(p.started) | ||
|
|
||
| return 0, err | ||
| } | ||
|
|
||
| p.Started <- struct{}{} | ||
| // Broadcast that the process has successfully started | ||
| p.startOnce.Do(func() { | ||
| close(p.started) | ||
| }) |
Comment on lines
+154
to
+167
| // Wait for the upstream process to either start successfully or fail immediately. | ||
| // This prevents signaling before the process is fully initialized. | ||
| select { | ||
| case result := <-resultChan: | ||
| slog.Info("Upstream process terminated before signal could be sent.") | ||
| return resolveExitCode(result, fallbackExitCode) | ||
| case <-upstream.Started(): | ||
| // Process is running (or failed to start), proceed | ||
| } | ||
|
|
||
| slog.Info("Sending signal to upstream process...", "signal", sig) | ||
| if err := upstream.Signal(sig); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } |
Comment on lines
+195
to
+200
| func (s *Service) isPortOpen() bool { | ||
| portStr := strconv.Itoa(s.config.TargetPort) | ||
| addrs := []string{ | ||
| net.JoinHostPort("127.0.0.1", portStr), | ||
| net.JoinHostPort("::1", portStr), | ||
| } |
Comment on lines
+58
to
62
| if exitErr, ok := err.(*exec.ExitError); ok { | ||
| return exitCodeFromProcessState(exitErr.ProcessState), nil | ||
| } | ||
|
|
||
| return 0, err |
Comment on lines
+241
to
+244
| // Attempt a fast TCP connection immediately to prevent unnecessary latency on boot | ||
| if tryDial() { | ||
| return nil, nil, nil | ||
| } |
Comment on lines
+164
to
172
| slog.Info("Sending signal to upstream process...", "signal", sig) | ||
| if err := upstream.Signal(sig); err != nil { | ||
| slog.Error("Failed to send signal to upstream process", "error", err) | ||
| } | ||
|
|
||
| exitCode, err := upstream.Run() | ||
| if err != nil { | ||
| slog.Error("Failed to start wrapped process", "command", s.config.UpstreamCommand, "args", s.config.UpstreamArgs, "error", err) | ||
| return 1 | ||
| if res, ok := waitResult(resultChan, timeout); ok { | ||
| slog.Info("Upstream process terminated after signal.") | ||
| return resolveExitCode(res, fallbackExitCode) | ||
| } |
Comment on lines
323
to
+326
| func (s *Service) setEnvironment() { | ||
| // Set PORT to be inherited by the upstream process. | ||
| os.Setenv("PORT", fmt.Sprintf("%d", s.config.TargetPort)) | ||
| os.Setenv("PORT", strconv.Itoa(s.config.TargetPort)) | ||
| } |
Comment on lines
+25
to
+27
| func (p *UpstreamProcess) Started() <-chan struct{} { | ||
| return p.started | ||
| } |
Comment on lines
34
to
+39
| err := p.cmd.Start() | ||
|
|
||
| // Broadcast that the start attempt has concluded (unblocks waiters on success OR failure) | ||
| p.startOnce.Do(func() { | ||
| close(p.started) | ||
| }) |
Comment on lines
+20
to
+22
| UpstreamCommand: os.Args[0], | ||
| UpstreamArgs: []string{"-test.run=TestHelperProcess"}, | ||
| TargetPort: 3000, |
Comment on lines
+58
to
+59
| p := NewUpstreamProcess(os.Args[0], "-test.run=TestUpstreamHelperProcess") | ||
| p.cmd.Env = append(os.Environ(), "GO_WANT_UPSTREAM_HELPER_PROCESS=1") |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR introduces a simpler, zero-config alternative to HTTP-based health checks by adding an opt-in TCP port readiness check - Thruster can now simply wait for the upstream process to bind to its target TCP port before booting the main proxy server. Added
WAIT_FOR_TARGET_PORT(default: false) andWAIT_FOR_TARGET_PORT_TIMEOUT(default: 60s) environment variables.