From b8ebaffd2e5b597b29b7e54239ca4e150c0c254a Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 7 Nov 2026 16:10:12 +0800 Subject: [PATCH] fix(restore): name the server that keeps native restore unavailable The native restore gate refused a POST restore and a queued restore with one message, "unavailable during a rolling upgrade: not every server of the cluster supports it yet", whether a server answered another protocol or was simply not connected. A cluster with one server down thus reported a rolling upgrade, and the operator looked for an older binary instead of the server. The gate now returns an error that wraps errNativeRestoreUnavailable and says why: errNativeRestorePeerUnreachable names the server that is not connected ("server host:port is not reachable, so it cannot be shown to have stopped an older restore"), errNativeRestoreUnsupported keeps the rolling-upgrade text for a server that reports another protocol or reconnected while it was asked, and a check that its deadline ended carries that error. The POST handler maps the error to 503 ServiceUnavailable with the message as before, the worker still clears its marker and stops, and errors.Is(err, errNativeRestoreUnavailable) holds for every refusal. The gate remains closed in every case it was closed before. TestNativeRestoreRefusalReasons posts a restore with a fake server that is not connected and one that answers protocol 1; before this change the first answered 503 with the rolling-upgrade text and without "not reachable". Co-Authored-By: Claude Fable 5.1 Signed-off-by: Feng Ruohang --- cmd/restore-capability.go | 73 +++++++++++++++++++++++++++------- cmd/restore-capability_test.go | 61 +++++++++++++++++++++++++++- 2 files changed, 117 insertions(+), 17 deletions(-) diff --git a/cmd/restore-capability.go b/cmd/restore-capability.go index ce84968da..09e438420 100644 --- a/cmd/restore-capability.go +++ b/cmd/restore-capability.go @@ -55,7 +55,34 @@ const ( restoreCapabilityTimeout = 5 * time.Second ) -var errNativeRestoreUnavailable = errors.New("native restore is unavailable during a rolling upgrade: not every server of the cluster supports it yet") +// errNativeRestoreUnavailable is the refusal of a native restore. The gate +// returns an error that wraps it and names the reason, so that an operator +// looks for a server that is not connected, not for an older binary. +var errNativeRestoreUnavailable = errors.New("native restore is unavailable") + +var ( + // errNativeRestoreUnsupported: a server of the cluster does not report + // this protocol, as an older server does not, or reconnected while it + // was asked, so its answer does not count. + errNativeRestoreUnsupported = fmt.Errorf("%w during a rolling upgrade: not every server of the cluster supports it yet", errNativeRestoreUnavailable) + // errNativeRestorePeerUnreachable: a server of the cluster is not + // connected. nativeRestoreUnreachableError names it. + errNativeRestorePeerUnreachable = fmt.Errorf("%w: a server of the cluster is not reachable", errNativeRestoreUnavailable) +) + +// nativeRestoreUnreachableError is errNativeRestorePeerUnreachable naming +// the server. +type nativeRestoreUnreachableError struct{ server string } + +func (e nativeRestoreUnreachableError) Error() string { + server := "a server of the cluster" + if e.server != "" { + server = "server " + e.server + } + return fmt.Sprintf("native restore is unavailable: %s is not reachable, so it cannot be shown to have stopped an older restore", server) +} + +func (nativeRestoreUnreachableError) Unwrap() error { return errNativeRestorePeerUnreachable } // restoreCapabilityHandler answers the question on this server. func restoreCapabilityHandler(*grid.MSS) (*grid.MSS, *grid.RemoteErr) { @@ -64,6 +91,8 @@ func restoreCapabilityHandler(*grid.MSS) (*grid.MSS, *grid.RemoteErr) { // restoreCapabilityPeer is another server of the cluster. type restoreCapabilityPeer interface { + // restoreServer names the server, for messages; empty when unknown. + restoreServer() string // restoreConnection identifies the current connection to the server; it // changes when the server reconnects, as a restarted one does. ok is // false when the server is not connected. @@ -73,6 +102,13 @@ type restoreCapabilityPeer interface { type restorePeerClient struct{ *peerRESTClient } +func (c restorePeerClient) restoreServer() string { + if c.peerRESTClient == nil || c.host == nil { + return "" + } + return c.host.String() +} + func (c restorePeerClient) restoreConnection() (string, bool) { if c.peerRESTClient == nil || c.gridConn == nil { return "", false @@ -126,8 +162,9 @@ func notificationRestorePeers() []restoreCapabilityPeer { return peers } -// nativeRestoreAvailable returns errNativeRestoreUnavailable unless every -// other server of the cluster supports native restore. +// nativeRestoreAvailable returns an error that wraps +// errNativeRestoreUnavailable, and says why, unless every other server of +// the cluster supports native restore. func nativeRestoreAvailable(ctx context.Context) error { return globalRestoreCapability.check(ctx) } @@ -137,34 +174,35 @@ func (g *restoreCapabilityGate) check(ctx context.Context) error { // One deadline covers the wait for another check and the questions. ctx, cancel := context.WithTimeout(ctx, restoreCapabilityTimeout) defer cancel() + ended := func() error { return fmt.Errorf("%w: %w", errNativeRestoreUnavailable, ctx.Err()) } select { case g.turn <- struct{}{}: case <-ctx.Done(): - return errNativeRestoreUnavailable + return ended() } defer func() { <-g.turn }() // Read the connections only now: one may have changed while this check - // waited for another. + // waited for another. A server that is not connected is named. peers := g.peers() - connections := func() ([]string, bool) { + connections := func() ([]string, error) { keys := make([]string, len(peers)) for i, p := range peers { key, ok := p.restoreConnection() if !ok { - return nil, false + return nil, nativeRestoreUnreachableError{server: p.restoreServer()} } keys[i] = key } - return keys, true + return keys, nil } - keys, ok := connections() - if !ok { + keys, err := connections() + if err != nil { g.checked, g.keys = time.Time{}, nil - return errNativeRestoreUnavailable + return err } if ctx.Err() != nil { - return errNativeRestoreUnavailable + return ended() } if !g.checked.IsZero() && time.Since(g.checked) < restoreCapabilityTTL && slices.Equal(keys, g.keys) { return nil @@ -183,9 +221,14 @@ func (g *restoreCapabilityGate) check(ctx context.Context) error { wg.Wait() // An answer counts only for the connection it was asked on: a server // that reconnected meanwhile may be another binary. - after, ok := connections() - if !ok || !slices.Equal(keys, after) || slices.Contains(answers, false) || ctx.Err() != nil { - return errNativeRestoreUnavailable + after, err := connections() + switch { + case err != nil: + return err + case ctx.Err() != nil: + return ended() + case !slices.Equal(keys, after), slices.Contains(answers, false): + return errNativeRestoreUnsupported } g.checked, g.keys = time.Now(), keys return nil diff --git a/cmd/restore-capability_test.go b/cmd/restore-capability_test.go index 306764d12..49840344a 100644 --- a/cmd/restore-capability_test.go +++ b/cmd/restore-capability_test.go @@ -74,21 +74,26 @@ func TestNativeRestoreCapabilityGrid(t *testing.T) { } } -// restoreCapabilityFake is a server whose connection and answer a test sets. +// restoreCapabilityFake is a server whose connection and answer a test sets: +// capable answers the question, protocol, when set, with another protocol. type restoreCapabilityFake struct { + server string capable atomic.Bool connected atomic.Bool conn atomic.Int64 asked atomic.Int64 + protocol atomic.Pointer[string] } func newRestoreCapabilityFake() *restoreCapabilityFake { - p := &restoreCapabilityFake{} + p := &restoreCapabilityFake{server: "server-2:9000"} p.capable.Store(true) p.connected.Store(true) return p } +func (p *restoreCapabilityFake) restoreServer() string { return p.server } + func (p *restoreCapabilityFake) restoreConnection() (string, bool) { return strings.Repeat("c", int(p.conn.Load())), p.connected.Load() } @@ -98,6 +103,9 @@ func (p *restoreCapabilityFake) restoreCapability(context.Context) (string, erro if !p.capable.Load() { return "", errors.New("Invalid Handler for type: RestoreCapability") } + if protocol := p.protocol.Load(); protocol != nil { + return *protocol, nil + } return nativeRestoreCapability, nil } @@ -186,6 +194,55 @@ func TestNativeRestoreRollingUpgradeGate(t *testing.T) { } } +// A refused POST restore says why. A server that is not connected is named, +// as not reachable: that is not a rolling upgrade, and the operator looks for +// the server. A server that speaks another protocol is one. Neither refusal +// writes anything. +func TestNativeRestoreRefusalReasons(t *testing.T) { + r := newRestoreSafetyTest(t, 1, versioning.Enabled) + initConfigSubsystem(t.Context(), r.z) + oi, _ := restoreIsolationSeed(t, r, 0, false, true) + p := newRestoreCapabilityFake() + previous := globalRestoreCapability + globalRestoreCapability = &restoreCapabilityGate{peers: func() []restoreCapabilityPeer { return []restoreCapabilityPeer{p} }} + t.Cleanup(func() { globalRestoreCapability = previous }) + before := r.fileInfo(oi.VersionID) + layer := newRestoreIsolationLayer(r) + + p.connected.Store(false) + rec := restoreIsolationPOST(t, r, layer, oi.VersionID) + if body := rec.Body.String(); rec.Code != http.StatusServiceUnavailable || !strings.Contains(body, "not reachable") || + !strings.Contains(body, p.server) || strings.Contains(body, "rolling upgrade") { + t.Errorf("POST with a server not connected: %d %s", rec.Code, body) + } + if err := nativeRestoreAvailable(t.Context()); !errors.Is(err, errNativeRestorePeerUnreachable) || + !errors.Is(err, errNativeRestoreUnavailable) || errors.Is(err, errNativeRestoreUnsupported) { + t.Errorf("gate with a server not connected: %v", err) + } + + p.connected.Store(true) + older := "1" + p.protocol.Store(&older) + rec = restoreIsolationPOST(t, r, layer, oi.VersionID) + if body := rec.Body.String(); rec.Code != http.StatusServiceUnavailable || !strings.Contains(body, "rolling upgrade") || + strings.Contains(body, "not reachable") { + t.Errorf("POST with a server of protocol %s: %d %s", older, rec.Code, body) + } + if err := nativeRestoreAvailable(t.Context()); !errors.Is(err, errNativeRestoreUnsupported) || + !errors.Is(err, errNativeRestoreUnavailable) || errors.Is(err, errNativeRestorePeerUnreachable) { + t.Errorf("gate with a server of protocol %s: %v", older, err) + } + + if after := r.fileInfo(oi.VersionID); !reflect.DeepEqual(before.Metadata, after.Metadata) { + t.Error("a refused POST changed the version") + } + select { + case err := <-layer.done: + t.Fatalf("a refused POST started a restore: %v", err) + default: + } +} + // restoreCapabilityHeld is a fake server whose answer a test holds. type restoreCapabilityHeld struct { restoreCapabilityFake