mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-23 20:57:02 +00:00
fix: count only the first shutdown ACK from each node so a duplicate cannot stand in for a silent node (#1233)
This commit is contained in:
@@ -600,6 +600,8 @@ func drainDHCPLeases(nc *nats.Conn, timeout time.Duration) (released, responders
|
||||
// collectShutdownACKs publishes a shutdown request and collects ACKs from nodes.
|
||||
// When nodeFilter is non-empty, only ACKs from that node count toward nodeCount
|
||||
// and all others are ignored (used by the single-node local-drain path).
|
||||
// Only the first ACK from each node counts; repeats are dropped so a duplicate
|
||||
// cannot stand in for a node that never answered.
|
||||
func collectShutdownACKs(nc *nats.Conn, topic string, reqData []byte, nodeCount int, nodeFilter string, timeout time.Duration) ([]daemon.ShutdownACK, error) {
|
||||
inbox := nats.NewInbox()
|
||||
sub, err := nc.SubscribeSync(inbox)
|
||||
@@ -614,6 +616,7 @@ func collectShutdownACKs(nc *nats.Conn, topic string, reqData []byte, nodeCount
|
||||
nc.Flush()
|
||||
|
||||
var acks []daemon.ShutdownACK
|
||||
seen := make(map[string]bool, nodeCount)
|
||||
deadline := time.Now().Add(timeout)
|
||||
for len(acks) < nodeCount {
|
||||
remaining := time.Until(deadline)
|
||||
@@ -631,6 +634,10 @@ func collectShutdownACKs(nc *nats.Conn, topic string, reqData []byte, nodeCount
|
||||
if nodeFilter != "" && ack.Node != nodeFilter {
|
||||
continue
|
||||
}
|
||||
if seen[ack.Node] {
|
||||
continue
|
||||
}
|
||||
seen[ack.Node] = true
|
||||
acks = append(acks, ack)
|
||||
}
|
||||
return acks, nil
|
||||
|
||||
@@ -541,3 +541,39 @@ func TestReportGuestsLeftRunningSevereLogsErrorAndNamesGuests(t *testing.T) {
|
||||
assert.Contains(t, out, "i-abandoned")
|
||||
assert.NotContains(t, out, "i-stopped")
|
||||
}
|
||||
|
||||
// shutdownACKResponder answers each request on topic with one ACK per entry
|
||||
// in nodes, so a repeated name simulates one node replying twice.
|
||||
func shutdownACKResponder(t *testing.T, nc *nats.Conn, topic string, nodes ...string) {
|
||||
t.Helper()
|
||||
sub, err := nc.Subscribe(topic, func(msg *nats.Msg) {
|
||||
for _, node := range nodes {
|
||||
body, _ := json.Marshal(daemon.ShutdownACK{Node: node, Phase: "gate"})
|
||||
_ = msg.Respond(body)
|
||||
}
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
}
|
||||
|
||||
func TestCollectShutdownACKsCountsADuplicateNodeOnce(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
shutdownACKResponder(t, nc, "test.shutdown.dup", "node1", "node1")
|
||||
|
||||
acks, err := collectShutdownACKs(nc, "test.shutdown.dup", []byte(`{}`), 2, "", 300*time.Millisecond)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, acks, 1, "a second ACK from node1 must not stand in for the silent node2")
|
||||
assert.Equal(t, "node1", acks[0].Node)
|
||||
}
|
||||
|
||||
func TestCollectShutdownACKsNodeFilterIgnoresOtherNodes(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
shutdownACKResponder(t, nc, "test.shutdown.filter", "node2", "node1", "node1")
|
||||
|
||||
acks, err := collectShutdownACKs(nc, "test.shutdown.filter", []byte(`{}`), 1, "node1", 300*time.Millisecond)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, acks, 1)
|
||||
assert.Equal(t, "node1", acks[0].Node)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user