fix(cli): roll back an upgrade whose new build never becomes ready

- Stop closes the control server right after the deactivation-pin check,
  so a stopping process no longer answers /started for its successor.
  releaseResources still closes it for stop paths that skip Stop; a
  sync.Once makes the double close safe.
- upgrade and restart wait for a 200 from /started via
  newReadySocketControlClient; a 408 (start never finished) now rolls the
  upgrade back and makes restart warn instead of reporting success.
- The plain socket client still accepts any answer, for the
  deactivation-pin path.
This commit is contained in:
Dev Scribe authored and Cuong Manh Le committed 2026-10-08 21:45:17 +07:00
1 parent be8c068fa9
commit f391318ebc
5 files changed
+225 -11

No files matched your search

+38 -3
View File
@@ -2355,7 +2355,37 @@ func newSocketControlClient(ctx context.Context, s service.Service, dir string)
// newSocketControlClientWithTimeout returns new control client after control server was started.
// The timeoutDuration controls how long to wait for the server.
//
// Any answer to /started counts, including the 408 of a service whose start
// never finished: callers such as the deactivation pin check need to reach a
// running service whether or not it is ready. A caller that needs it ready uses
// newReadySocketControlClient.
func newSocketControlClientWithTimeout(ctx context.Context, s service.Service, dir string, timeoutDuration time.Duration) *controlClient {
return dialSocketControlServer(ctx, s, dir, timeoutDuration, false)
}
// newReadySocketControlClient is newSocketControlClient for a caller that needs
// the service ready, not just reachable: only a 200 from /started counts.
//
// A service whose start never finishes still runs its control server, and
// answers /started with 408 once its onStarted hooks have not completed in
// time. `ctrld upgrade` must read that as "the new binary did not become
// ready" and roll back, rather than report the upgrade successful.
func newReadySocketControlClient(ctx context.Context, s service.Service, dir string) *controlClient {
return newReadySocketControlClientWithTimeout(ctx, s, dir, dialSocketControlServerTimeout)
}
// newReadySocketControlClientWithTimeout is newReadySocketControlClient with
// an explicit deadline.
func newReadySocketControlClientWithTimeout(ctx context.Context, s service.Service, dir string, timeoutDuration time.Duration) *controlClient {
return dialSocketControlServer(ctx, s, dir, timeoutDuration, true)
}
// dialSocketControlServer polls /started until the control server answers, or
// with requireReady until it answers 200, and returns the client. It returns
// nil if the service is not running, or the answer does not come within
// timeoutDuration.
func dialSocketControlServer(ctx context.Context, s service.Service, dir string, timeoutDuration time.Duration, requireReady bool) *controlClient {
// Return early if service is not running.
if status, err := s.Status(); err != nil || status != service.StatusRunning {
return nil
@@ -2370,10 +2400,15 @@ func newSocketControlClientWithTimeout(ctx context.Context, s service.Service, d
// The socket control server may not start yet, so attempt to ping
// it until we got a response.
for {
_, err := cc.post(startedPath, nil)
resp, err := cc.post(startedPath, nil)
if err == nil {
// Server was started, stop pinging.
break
code := resp.StatusCode
resp.Body.Close()
if !requireReady || code == http.StatusOK {
// Server was started, stop pinging.
break
}
err = fmt.Errorf("control server reported the service not ready: HTTP %d", code)
}
// The socket control server is not ready yet, backoff for waiting it to be ready.
bo.BackOff(ctx, err)
+8 -2
View File
@@ -92,14 +92,20 @@ func (sc *ServiceCommand) Restart(cmd *cobra.Command, args []string) error {
if validateConfigErr != nil {
timeout = 5 * time.Second
}
if cc := newSocketControlClientWithTimeout(context.TODO(), s, dir, timeout); cc != nil {
// Only a ready service counts. Restart replaces nothing, so
// there is nothing to roll back, but a service that came
// back without finishing its start must not be reported as
// restarted cleanly.
if cc := newReadySocketControlClientWithTimeout(context.TODO(), s, dir, timeout); cc != nil {
_, _ = cc.post(ifacePath, nil)
logger.Debug().Msg("Control server ping successful")
} else {
logger.Warn().Err(err).Msg("Service was restarted, but ctrld process may not be ready yet")
logger.Warn().Msgf("Service was restarted, but ctrld did not report ready within %s; check `ctrld status`", timeout)
return nil
}
} else {
logger.Warn().Err(err).Msg("Service was restarted, but could not ping the control server")
return nil
}
logger.Notice().Msg("Service restarted")
} else {
+3 -1
View File
@@ -125,7 +125,9 @@ func (uc *UpgradeCommand) Upgrade(cmd *cobra.Command, args []string) error {
}
if doTasks(tasks) {
if dir, err := socketDir(); err == nil {
if cc := newSocketControlClient(context.TODO(), s, dir); cc != nil {
// Only a ready service counts: a build that starts but
// never becomes ready must be rolled back.
if cc := newReadySocketControlClient(context.TODO(), s, dir); cc != nil {
_, _ = cc.post(ifacePath, nil)
return true
}
+26 -5
View File
@@ -389,6 +389,7 @@ type prog struct {
restoreOnce sync.Once
restoreErr error
releaseOnce sync.Once
csStopOnce sync.Once
runDone chan struct{}
runAbortCh chan struct{}
listenerWg sync.WaitGroup
@@ -1407,11 +1408,7 @@ func (p *prog) restoreOSState() error {
// queries on would fail those queries. Safe to call multiple times.
func (p *prog) releaseResources() {
p.releaseOnce.Do(func() {
if p.cs != nil {
if cerr := p.cs.stop(); cerr != nil {
p.Warn().Err(cerr).Msg("could not stop control server")
}
}
p.stopControlServer()
p.closeLogConn()
p.mu.Lock()
upstreams := p.cfg.Upstream
@@ -1420,6 +1417,27 @@ func (p *prog) releaseResources() {
})
}
// stopControlServer closes the control socket. Stop calls it as soon as the stop
// is accepted, and releaseResources again for the stop paths that never call
// Stop. Safe to call multiple times.
//
// It must not wait for the run to finish. A stopping process that still answers
// on the socket answers for the service that replaces it: `ctrld upgrade` and
// restart take any reply to /started as "the new binary is up", so an upgrade
// to a build that never became ready was reported successful instead of rolled
// back. Closing late also unlinks the socket path after the new process has
// bound its own there.
func (p *prog) stopControlServer() {
p.csStopOnce.Do(func() {
if p.cs == nil {
return
}
if err := p.cs.stop(); err != nil {
p.Warn().Err(err).Msg("could not stop control server")
}
})
}
// shutdown runs the full teardown, in the order both halves require. It is
// called once the listeners stopped, on every stop path including the mobile
// one where the OS never terminates the process.
@@ -1479,6 +1497,9 @@ func (p *prog) Stop(_ service.Service) error {
os.Exit(deactivationPinInvalidExitCode)
}
}
// The pin check above is the last use of the control server. From here on
// this process is going away and must not answer for its successor.
p.stopControlServer()
close(p.stopCh)
return err
}
+150
View File
@@ -0,0 +1,150 @@
package cli
import (
"context"
"net/http"
"os"
"path/filepath"
"sync/atomic"
"testing"
"time"
"github.com/kardianos/service"
"github.com/stretchr/testify/require"
"github.com/Control-D-Inc/ctrld"
)
// A stopping process must not answer /started for the service that replaces
// it. `ctrld upgrade` treats any reply as "the new binary is up", so a reply
// from the old process, while it waits for its components to stop, turned an
// upgrade to a never-ready build into "Upgrade successful" instead of a
// rollback (ctrld-qa upgrade-readiness-rollback on v1.0).
func TestStopClosesControlServerBeforeRunFinishes(t *testing.T) {
// A short path, as the other control server tests use: t.TempDir() is
// longer than a Unix socket path may be.
f, err := os.CreateTemp("", "")
require.NoError(t, err)
sock := f.Name()
f.Close()
t.Cleanup(func() { os.Remove(sock) })
cs, err := newControlServer(sock)
require.NoError(t, err)
cs.register(startedPath, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
require.NoError(t, cs.start())
p := &prog{
cfg: &ctrld.Config{},
cs: cs,
waitCh: make(chan struct{}), stopCh: make(chan struct{}), runDone: make(chan struct{}),
runAbortCh: make(chan struct{}), dnsWatcherStopCh: make(chan struct{}),
pinCodeValidCh: make(chan struct{}, 1),
}
p.logger.Store(mainLog.Load())
p.pinCodeValidCh <- struct{}{} // accept the stop even if a deactivation pin is configured
cc := newControlClient(sock)
resp, err := cc.post(startedPath, nil)
require.NoError(t, err, "the running service answers")
resp.Body.Close()
require.NoError(t, p.Stop(nil))
// runDone is still open: the run has not finished, as when a component is
// slow to stop. The socket must already be closed.
_, err = cc.post(startedPath, nil)
require.Error(t, err, "a stopped service answered /started")
p.stopControlServer() // the later releaseResources call is harmless
}
// startTestControlServer starts a control server on a short socket path in a
// directory of its own, as the socket client expects, answering /started with
// whatever started returns.
func startTestControlServer(t *testing.T, started func() int) (dir, sock string, cs *controlServer) {
t.Helper()
// A short path, as the other control server tests use: t.TempDir() is
// longer than a Unix socket path may be.
dir, err := os.MkdirTemp("", "cs")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(dir) })
sock = filepath.Join(dir, ctrldControlUnixSock)
cs, err = newControlServer(sock)
require.NoError(t, err)
cs.register(startedPath, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(started()) }))
cs.register(listClientsPath, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
require.NoError(t, cs.start())
t.Cleanup(func() { _ = cs.stop() })
// start serves on a goroutine, and the server tracks the listener only
// once that runs: a Shutdown before then leaves the listener, and its
// unlink, to Serve's own deferred close. Wait until it answers, as a
// service that has been running does. A path other than /started keeps
// the started func's call count to the test's own requests.
require.Eventually(t, func() bool {
resp, err := newControlClient(sock).post(listClientsPath, nil)
if err != nil {
return false
}
resp.Body.Close()
return true
}, 5*time.Second, 10*time.Millisecond, "control server did not start serving")
return dir, sock, cs
}
// Stop unlinks the socket path as it closes the socket, before the successor
// starts: unlinking it later removed the path the successor had bound.
func TestStopRemovesControlSocketPath(t *testing.T) {
_, sock, cs := startTestControlServer(t, func() int { return http.StatusOK })
p := &prog{
cfg: &ctrld.Config{},
cs: cs,
waitCh: make(chan struct{}), stopCh: make(chan struct{}), runDone: make(chan struct{}),
runAbortCh: make(chan struct{}), dnsWatcherStopCh: make(chan struct{}),
pinCodeValidCh: make(chan struct{}, 1),
}
p.logger.Store(mainLog.Load())
p.pinCodeValidCh <- struct{}{}
require.NoError(t, p.Stop(nil))
_, err := os.Stat(sock)
require.True(t, os.IsNotExist(err), "socket path still present after Stop: %v", err)
}
// A stop path that never calls Stop still closes the control server.
func TestReleaseResourcesClosesControlServer(t *testing.T) {
_, sock, cs := startTestControlServer(t, func() int { return http.StatusOK })
p := &prog{cfg: &ctrld.Config{}, cs: cs}
p.logger.Store(mainLog.Load())
p.releaseResources()
_, err := newControlClient(sock).post(startedPath, nil)
require.Error(t, err, "the control server answered after releaseResources")
}
// `ctrld upgrade` must not count a successor that started but never became
// ready: it answers /started with 408, and only a 200 means ready. The plain
// socket client keeps accepting any answer, because the deactivation pin check
// must reach a running service whether or not it is ready.
func TestReadySocketControlClientRequiresReady(t *testing.T) {
running := &fakeService{statuses: []service.Status{service.StatusRunning}}
t.Run("never ready", func(t *testing.T) {
dir, _, _ := startTestControlServer(t, func() int { return http.StatusRequestTimeout })
require.Nil(t, newReadySocketControlClientWithTimeout(context.Background(), running, dir, 2*time.Second),
"a service answering 408 counted as ready")
require.NotNil(t, newSocketControlClientWithTimeout(context.Background(), running, dir, 2*time.Second),
"the plain client must still reach a running service that is not ready")
})
t.Run("becomes ready", func(t *testing.T) {
var calls atomic.Int32
dir, _, _ := startTestControlServer(t, func() int {
if calls.Add(1) == 1 {
return http.StatusRequestTimeout
}
return http.StatusOK
})
require.NotNil(t, dialSocketControlServer(context.Background(), running, dir, 10*time.Second, true),
"the poll gave up on a service that became ready")
require.GreaterOrEqual(t, calls.Load(), int32(2), "the poll stopped at the 408")
})
}