storage: fix decommissioning of absent node

When a node is decommissioned in absentia, the gateway node may not be
aware of its most recent liveness entry. Before this bug fix, the
gateway would fail the decommissioning process.  Instead, it now falls
back to reading the liveness record from the KV store.

Fixes #17995.

Release note (bug fix): Decommissioning a node that has already been
terminated now works in all cases. Success previously depended on
whether the gateway node "remembered" the absent decommissionee.

pkg/storage/node_liveness.go

@@ -256,7 +256,33 @@ func (nl *NodeLiveness) SetDecommissioning(
 			<-sem
 		}()
 
-		oldLiveness, err := nl.GetLiveness(nodeID) // need new liveness in each iteration
+		// We need the current liveness in each iteration.
+		//
+		// We ignore any liveness record in Gossip because we may have to fall back
+		// to the KV store anyway. The scenario in which this is needed is:
+		// - kill node 2 and stop node 1
+		// - wait for node 2's liveness record's Gossip entry to expire on all surviving nodes
+		// - restart node 1; it'll never see node 2 in `GetLiveness` unless the whole
+		//   node liveness span gets regossiped (unlikely if it wasn't the lease holder
+		//   for that span)
+		// - can't decommission node 2 from node 1 without KV fallback.
+		//
+		// See #20863.
+		//
+		// NB: this also de-flakes TestNodeLivenessDecommissionAbsent; running
+		// decommissioning commands in a tight loop on different nodes sometimes
+		// results in unintentional no-ops (due to the Gossip lag); this could be
+		// observed by users in principle, too.
+		dummyLiveness := &Liveness{NodeID: nodeID} // always catches condition failed
+		var oldLiveness *Liveness
+		err := nl.updateLiveness(ctx, dummyLiveness, dummyLiveness, func(actual Liveness) error {
+			if (actual == Liveness{}) {
+				return ErrNoLivenessRecord
+			}
+			oldLiveness = &actual
+			return nil
+		})
+
 		if err != nil {
 			return false, errors.Wrap(err, "unable to get liveness")
 		}

pkg/storage/node_liveness_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/pkg/errors"
 
 	"github.com/cockroachdb/cockroach/pkg/gossip"
@@ -797,6 +798,70 @@ func TestNodeLivenessSetDecommissioning(t *testing.T) {
 	testNodeLivenessSetDecommissioning(t, 1)
 }
 
+// TestNodeLivenessDecommissionAbsent exercises a scenario in which a node is
+// asked to decommission another node whose liveness record is not gossiped any
+// more.
+//
+// See (*NodeLiveness).SetDecommissioning for details.
+func TestNodeLivenessDecommissionAbsent(t *testing.T) {
+	mtc := &multiTestContext{}
+	defer mtc.Stop()
+	mtc.Start(t, 3)
+	mtc.initGossipNetwork()
+
+	verifyLiveness(t, mtc)
+
+	ctx := context.Background()
+	const goneNodeID = roachpb.NodeID(10000)
+
+	// When the node simply never existed, expect an error.
+	if _, err := mtc.nodeLivenesses[0].SetDecommissioning(
+		ctx, goneNodeID, true,
+	); errors.Cause(err) != storage.ErrNoLivenessRecord {
+		t.Fatal(err)
+	}
+
+	// Pretend the node was once there but isn't gossiped anywhere.
+	if err := mtc.dbs[0].CPut(ctx, keys.NodeLivenessKey(goneNodeID), &storage.Liveness{
+		NodeID:     goneNodeID,
+		Epoch:      1,
+		Expiration: hlc.LegacyTimestamp(mtc.clock.Now()),
+	}, nil); err != nil {
+		t.Fatal(err)
+	}
+
+	// Decommission from second node.
+	if committed, err := mtc.nodeLivenesses[1].SetDecommissioning(ctx, goneNodeID, true); err != nil {
+		t.Fatal(err)
+	} else if !committed {
+		t.Fatal("no change committed")
+	}
+	// Re-decommission from first node.
+	if committed, err := mtc.nodeLivenesses[0].SetDecommissioning(ctx, goneNodeID, true); err != nil {
+		t.Fatal(err)
+	} else if committed {
+		t.Fatal("spurious change committed")
+	}
+	// Recommission from first node.
+	if committed, err := mtc.nodeLivenesses[0].SetDecommissioning(ctx, goneNodeID, false); err != nil {
+		t.Fatal(err)
+	} else if !committed {
+		t.Fatal("no change committed")
+	}
+	// Decommission from second node (a second time).
+	if committed, err := mtc.nodeLivenesses[1].SetDecommissioning(ctx, goneNodeID, true); err != nil {
+		t.Fatal(err)
+	} else if !committed {
+		t.Fatal("no change committed")
+	}
+	// Recommission from third node.
+	if committed, err := mtc.nodeLivenesses[2].SetDecommissioning(ctx, goneNodeID, false); err != nil {
+		t.Fatal(err)
+	} else if !committed {
+		t.Fatal("no change committed")
+	}
+}
+
 func TestNodeLivenessLivenessStatus(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	now := timeutil.Now()