From 90a13845d10f3300b20ff32237e998822694e017 Mon Sep 17 00:00:00 2001 From: Harsh Rawat Date: Tue, 11 Aug 2026 14:46:51 +0530 Subject: [PATCH] [live-migration] Fix source resume hang when cancelled before blackout On a source rollback, Resume unconditionally re-armed the log listener and re-accepted the guest GCS bridge, assuming a blackout had dropped them. When the migration is cancelled before blackout (e.g. cancel after PrepareAndExport), those connections are still live, so both accepts block until the GCS connection timeout. finalize(resume) then fails with DeadlineExceeded, leaving the session Cancelled and blocking cleanup. We now track whether the bridge transport is still live and, on resume, reuse the existing bridge and log stream instead of re-accepting when it never dropped. Signed-off-by: Harsh Rawat --- internal/controller/vm/save_lcow.go | 75 ++++++++++++++--------------- internal/gcs/bridge.go | 24 ++++++--- internal/gcs/guestconnection.go | 8 +++ internal/vm/guestmanager/guest.go | 11 +++++ 4 files changed, 72 insertions(+), 46 deletions(-) diff --git a/internal/controller/vm/save_lcow.go b/internal/controller/vm/save_lcow.go index 82916f3215..c7be49eca6 100644 --- a/internal/controller/vm/save_lcow.go +++ b/internal/controller/vm/save_lcow.go @@ -221,47 +221,46 @@ func (c *Controller) Resume(ctx context.Context, rebuildBridge bool) error { return fmt.Errorf("cannot resume from migration: VM is in state %s: %w", c.vmState, errdefs.ErrFailedPrecondition) } - // On the destination, the log connection was never established. - // On source, the blackout dropped the source's GCS log connection, which tore - // down its listener and closed logOutputDone. Install a fresh signal and - // re-arm the listener so the resumed guest's reconnect-mode vsockexec can - // reconnect and host-side logs resume. - c.logOutputDone = make(chan struct{}) - // We expect the reconnect to complete within the GCS connection timeout, - // otherwise we want to fail. - ctx, cancel := context.WithTimeout(ctx, timeout.GCSConnectionTimeout) - log.G(ctx).Debugf("using gcs connection timeout: %s\n", timeout.GCSConnectionTimeout) - - g, gctx := errgroup.WithContext(ctx) - defer func() { - _ = g.Wait() - }() - defer cancel() - - if err := c.setupLoggingListener(gctx, g); err != nil { - return fmt.Errorf("arm logging listener on resume: %w", err) - } - - if rebuildBridge { - // Source rollback: arm the host GCS listener now, then accept the guest's - // post-blackout re-dial and swap it into the running bridge. - if err := c.guest.PrepareConnection(winio.VsockServiceID(prot.LinuxGcsVsockPort)); err != nil { - return fmt.Errorf("prepare source resume listener: %w", err) - } - if err := c.guest.ResumeConnection(ctx); err != nil { - return fmt.Errorf("resume source guest connection: %w", err) + // A source rollback before blackout never dropped the guest connection, so the + // live bridge and its log stream are reused instead of re-accepted. + reuseLiveConn := rebuildBridge && c.guest.IsBridgeConnected() + + // Reuse skips the work below. Otherwise re-arm host logging and (re)establish the + // bridge, bounded by the GCS connection timeout; the shared tail keeps the caller ctx. + if !reuseLiveConn { + timeoutCtx, cancel := context.WithTimeout(ctx, timeout.GCSConnectionTimeout) + log.G(timeoutCtx).Debugf("using gcs connection timeout: %s\n", timeout.GCSConnectionTimeout) + g, gctx := errgroup.WithContext(timeoutCtx) + defer func() { + _ = g.Wait() + }() + defer cancel() + + // Logs were dropped by the source blackout and never established back. + c.logOutputDone = make(chan struct{}) + if err := c.setupLoggingListener(gctx, g); err != nil { + return fmt.Errorf("arm logging listener on resume: %w", err) } - } else { - // Destination: reuse the connection already armed at start. - if err := c.guest.CreateConnection(ctx, false); err != nil { - return fmt.Errorf("resume destination guest connection: %w", err) + + if rebuildBridge { + // Source rollback after blackout: accept the guest's re-dial into the bridge. + if err := c.guest.PrepareConnection(winio.VsockServiceID(prot.LinuxGcsVsockPort)); err != nil { + return fmt.Errorf("prepare source resume listener: %w", err) + } + if err := c.guest.ResumeConnection(timeoutCtx); err != nil { + return fmt.Errorf("resume source guest connection: %w", err) + } + } else { + // Destination: reuse the connection already armed at start. + if err := c.guest.CreateConnection(timeoutCtx, false); err != nil { + return fmt.Errorf("resume destination guest connection: %w", err) + } } - } - // Collect any errors from establishing the log connection. - // If the connection is not established then we need to error out. - if err := g.Wait(); err != nil { - return err + // Fail if the log connection could not be established. + if err := g.Wait(); err != nil { + return err + } } // Clear migrating flag only now that the new transport is in place. diff --git a/internal/gcs/bridge.go b/internal/gcs/bridge.go index b18112e624..d9d0229f06 100644 --- a/internal/gcs/bridge.go +++ b/internal/gcs/bridge.go @@ -54,15 +54,21 @@ type bridge struct { rpcs map[int64]*rpc // conn is the transport carrying messages to and from the guest. // Held atomically because the send path reads it while a migration swaps it. - conn atomic.Value - rpcCh chan *rpc - notify notifyFunc - closed bool - log *logrus.Entry - brdgErr error - waitCh chan struct{} + conn atomic.Value + rpcCh chan *rpc + notify notifyFunc + closed bool + log *logrus.Entry + brdgErr error + waitCh chan struct{} + + // Migration related fields + // migrating tolerates transport drops during a live-migration window. migrating atomic.Bool - resumeCh chan struct{} + // resumeCh wakes the parked recv loop when a new transport is swapped in. + resumeCh chan struct{} + // connected is true while a live transport is present and being read. + connected atomic.Bool } var ErrBridgeClosed = fmt.Errorf("bridge closed: %w", net.ErrClosed) @@ -305,7 +311,9 @@ func (brdg *bridge) RPC(ctx context.Context, proc prot.RPCProc, req requestMessa func (brdg *bridge) recvLoopRoutine() { for { + brdg.connected.Store(true) err := brdg.recvLoop() + brdg.connected.Store(false) if !brdg.migrating.Load() { brdg.kill(err) diff --git a/internal/gcs/guestconnection.go b/internal/gcs/guestconnection.go index 11ee39e2ab..2a60cc23ec 100644 --- a/internal/gcs/guestconnection.go +++ b/internal/gcs/guestconnection.go @@ -244,6 +244,14 @@ func (gc *GuestConnection) SetMigrating(migrating bool) { gc.brdg.SetMigrating(migrating) } +// IsBridgeConnected reports whether a live bridge transport is currently installed. +func (gc *GuestConnection) IsBridgeConnected() bool { + if gc.brdg == nil { + return false + } + return gc.brdg.connected.Load() +} + // ResumeOnConn resumes the bridge after swaping the bridge // transport without dropping outstanding RPCs. func (gc *GuestConnection) ResumeOnConn(ctx context.Context, conn io.ReadWriteCloser) error { diff --git a/internal/vm/guestmanager/guest.go b/internal/vm/guestmanager/guest.go index 3c550b911c..30bc857dc8 100644 --- a/internal/vm/guestmanager/guest.go +++ b/internal/vm/guestmanager/guest.go @@ -225,6 +225,17 @@ func (gm *Guest) SetMigrating(migrating bool) { gm.gc.SetMigrating(migrating) } +// IsBridgeConnected reports whether a live bridge transport is currently installed. +func (gm *Guest) IsBridgeConnected() bool { + gm.mu.RLock() + defer gm.mu.RUnlock() + + if gm.gc == nil { + return false + } + return gm.gc.IsBridgeConnected() +} + // ResumeConnection accepts a fresh hvsock on the prepared listener and // swaps it into the existing GCS bridge, preserving in-flight RPCs. func (gm *Guest) ResumeConnection(ctx context.Context) error {