amadan

carlos / platform Public

cmd/moveprobe: one shared client for the control socket

Built once at boot when CARLOS_CONTROL_SOCKET is set, and each reply is
drained before close, so repeated arms reuse one connection instead of
leaving a transport and its idle connection behind per call. Review fix
round 1, item 4.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5fc8c50dcdde5c79076f7c8af2f83c2437acbea8 parent 51316d0
2 files changed, +42 −12
  • cmd/moveprobe/main.go +19 −10
  • cmd/moveprobe/main_test.go +23 −2
diff --git a/cmd/moveprobe/main.go b/cmd/moveprobe/main.go
index 0e4f9e625..c65d84f21 100644
--- a/cmd/moveprobe/main.go
+++ b/cmd/moveprobe/main.go
@@ -224,6 +224,10 @@ type app struct {
workerFailed int64
workerInARow int64
workerLastErr string
+
+ // control is the one client for the instance control socket, nil when
+ // the agent delivered none.
+ control *http.Client
}
func openDB(path string) (*sql.DB, error) {
@@ -284,6 +288,14 @@ func boot(ctx context.Context, dbPath, here, from, mode string) (*app, error) {
return nil, err
}
a.setupStore()
+ if sock := os.Getenv("CARLOS_CONTROL_SOCKET"); sock != "" {
+ a.control = &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ var d net.Dialer
+ return d.DialContext(ctx, "unix", sock)
+ },
+ }}
+ }
if a.storeErr == nil {
putCtx, cancel := context.WithTimeout(ctx, storeWait)
if err := a.putObject(putCtx); err != nil {
@@ -746,8 +758,7 @@ func (a *app) handleArmOnce(w http.ResponseWriter, r *http.Request) {
http.Error(w, "in: want whole seconds, at most 400 days", http.StatusBadRequest)
return
}
- control := os.Getenv("CARLOS_CONTROL_SOCKET")
- if control == "" {
+ if a.control == nil {
http.Error(w, "no control socket: CARLOS_CONTROL_SOCKET is not set", http.StatusServiceUnavailable)
return
}
@@ -776,18 +787,16 @@ func (a *app) handleArmOnce(w http.ResponseWriter, r *http.Request) {
}
req.Header.Set("Authorization", "Bearer "+token)
req.Header.Set("Content-Type", "application/json")
- client := &http.Client{Transport: &http.Transport{
- DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
- var d net.Dialer
- return d.DialContext(ctx, "unix", control)
- },
- }}
- resp, err := client.Do(req)
+ resp, err := a.control.Do(req)
if err != nil {
http.Error(w, "control socket: "+err.Error(), http.StatusBadGateway)
return
}
- defer resp.Body.Close()
+ // Drained before close, so the connection goes back for the next arm.
+ defer func() {
+ io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<16))
+ resp.Body.Close()
+ }()
reply, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
if resp.StatusCode/100 != 2 {
http.Error(w, fmt.Sprintf("control socket: %d %s", resp.StatusCode, strings.TrimSpace(string(reply))), http.StatusBadGateway)
diff --git a/cmd/moveprobe/main_test.go b/cmd/moveprobe/main_test.go
index 3100e2ebb..8d28ee449 100644
--- a/cmd/moveprobe/main_test.go
+++ b/cmd/moveprobe/main_test.go
@@ -108,7 +108,9 @@ type fixture struct {
mu sync.Mutex
auths []string
storeReq atomic.Int64
- starts int
+ // controlConns counts the connections the control socket accepted.
+ controlConns atomic.Int64
+ starts int
}
func newFixture(t *testing.T) *fixture {
@@ -143,7 +145,11 @@ func newFixture(t *testing.T) *fixture {
}
mux := http.NewServeMux()
schedule.Mount(mux, testHost, reg, func() (string, error) { return adminToken, nil }, time.Now)
- cs := &http.Server{Handler: mux}
+ cs := &http.Server{Handler: mux, ConnState: func(_ net.Conn, s http.ConnState) {
+ if s == http.StateNew {
+ f.controlConns.Add(1)
+ }
+ }}
go cs.Serve(ln)
t.Cleanup(func() { cs.Close() })
@@ -1202,3 +1208,18 @@ func TestReportNeverAnswersPartial(t *testing.T) {
}
})
}
+
+func TestArmOnceReusesOneControlConnection(t *testing.T) {
+ f := newFixture(t)
+ p := f.start(f.env)
+ p.wait("/ready", http.StatusOK)
+ for i := 0; i < 3; i++ {
+ if code, body, _ := p.do(http.MethodPost, "/arm-once?in=600", probeSecret, nil); code != http.StatusOK {
+ t.Fatalf("arm %d: %d %q", i+1, code, body)
+ }
+ }
+ if n := f.controlConns.Load(); n != 1 {
+ t.Errorf("three arms opened %d connections to the control socket, want 1", n)
+ }
+ p.stop()
+}