From 4758b6c647d28da18b46127d58314fc952f69c65 Mon Sep 17 00:00:00 2001 From: dawn Date: Fri, 28 Aug 2026 22:28:43 +0900 Subject: [PATCH] spindle/mill: remove executor quarantine Signed-off-by: dawn --- cmd/spindle/main.go | 35 +--- spindle/db/db.go | 8 + spindle/db/mill_tokens.go | 55 +---- spindle/db/mill_tokens_test.go | 92 ++++++--- spindle/mill/lease.go | 6 + spindle/mill/mill.go | 184 +++++++++++------ spindle/mill/mill_test.go | 357 +++++++++++++++++++++++++++++---- spindle/mill/restore.go | 3 +- spindle/mill/restore_test.go | 17 ++ spindle/mill/session.go | 33 +-- 10 files changed, 577 insertions(+), 213 deletions(-) diff --git a/cmd/spindle/main.go b/cmd/spindle/main.go index b97f50f2..dcfda8f3 100644 --- a/cmd/spindle/main.go +++ b/cmd/spindle/main.go @@ -221,7 +221,7 @@ func millCommand() *cli.Command { return err } w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) - fmt.Fprintln(w, "NAME\tCREATED\tEXPIRES\tLABELS\tQUARANTINE") + fmt.Fprintln(w, "NAME\tCREATED\tEXPIRES\tLABELS") for _, registration := range registrations { expires := "never" if registration.ExpiresAt != nil { @@ -234,42 +234,11 @@ func millCommand() *cli.Command { if labels == "" { labels = "-" } - quarantine := "-" - if registration.QuarantineReason != nil { - quarantine = *registration.QuarantineReason - if registration.QuarantinedAt != nil { - quarantine = *registration.QuarantinedAt + ": " + quarantine - } - } - fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", registration.Name, registration.CreatedAt, expires, labels, quarantine) + fmt.Fprintf(w, "%s\t%s\t%s\t%s\n", registration.Name, registration.CreatedAt, expires, labels) } return w.Flush() }, }, - { - Name: "unquarantine", - Usage: "allow a quarantined executor to reconnect", - ArgsUsage: "", - Flags: []cli.Flag{dbFlag}, - Action: func(ctx context.Context, cmd *cli.Command) error { - name := cmd.Args().First() - if name == "" { - return fmt.Errorf("usage: spindle mill executor unquarantine ") - } - d, err := openDB(ctx, cmd) - if err != nil { - return err - } - ok, err := d.ClearExecutorQuarantine(name) - if err != nil { - return err - } - if !ok { - return fmt.Errorf("no such executor identity %q", name) - } - return nil - }, - }, { Name: "reset-cursor", Usage: "reset the mill's acked stream position for an executor after mill state loss; takes effect on the executor's next reconnect", diff --git a/spindle/db/db.go b/spindle/db/db.go index e52437b8..bc508e5b 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -378,6 +378,14 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + // older generations select these fields during authentication, so clear rather than drop them + if err := orm.RunMigration(conn, logger, "mill-clear-executor-quarantine", func(tx *sql.Tx) error { + _, err := tx.Exec(`update mill_executors set quarantine_reason = null, quarantined_at = null`) + return err + }); err != nil { + return err + } + if err := orm.RunMigration(conn, logger, "jobs-trace-context", func(tx *sql.Tx) error { for _, column := range []string{"traceparent", "tracestate"} { var present int diff --git a/spindle/db/mill_tokens.go b/spindle/db/mill_tokens.go index e3fd3422..e4a82f4e 100644 --- a/spindle/db/mill_tokens.go +++ b/spindle/db/mill_tokens.go @@ -11,12 +11,10 @@ import ( // the raw token is never stored, only its hash type ExecutorRegistration struct { - Name string - CreatedAt string - ExpiresAt *time.Time - Labels []string - QuarantineReason *string - QuarantinedAt *string + Name string + CreatedAt string + ExpiresAt *time.Time + Labels []string } func (d *DB) CreateExecutorToken(tokenHash string, expiresAt *time.Time) error { @@ -92,23 +90,19 @@ func (d *DB) RegisterExecutor(name, tokenHash string, labels []string) error { func (d *DB) ResolveExecutorToken(tokenHash string) (string, []string, bool, error) { var name string - var expires sql.NullString - var labelsRaw, quarantineReason sql.NullString + var expires, labelsRaw sql.NullString err := d.QueryRow(` - select e.name, t.expires_at, e.labels, e.quarantine_reason + select e.name, t.expires_at, e.labels from mill_executors e join mill_executor_tokens t on t.token_hash = e.token_hash where e.token_hash = ? - `, tokenHash).Scan(&name, &expires, &labelsRaw, &quarantineReason) + `, tokenHash).Scan(&name, &expires, &labelsRaw) if err == sql.ErrNoRows { return "", nil, false, nil } if err != nil { return "", nil, false, err } - if quarantineReason.Valid { - return "", nil, false, nil - } exp, hasExpiry, err := parseExpiry(expires) if err != nil { return "", nil, false, err @@ -125,28 +119,6 @@ func (d *DB) ResolveExecutorToken(tokenHash string) (string, []string, bool, err return name, labels, true, nil } -func (d *DB) QuarantineExecutor(name, reason string) error { - _, err := d.Exec( - `update mill_executors - set quarantine_reason = ?, quarantined_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') - where name = ?`, - reason, name, - ) - return err -} - -func (d *DB) ClearExecutorQuarantine(name string) (bool, error) { - res, err := d.Exec( - `update mill_executors set quarantine_reason = null, quarantined_at = null where name = ?`, - name, - ) - if err != nil { - return false, err - } - n, err := res.RowsAffected() - return n > 0, err -} - func (d *DB) RevokeExecutorToken(name string) (bool, error) { res, err := d.Exec(` delete from mill_executor_tokens @@ -161,8 +133,7 @@ func (d *DB) RevokeExecutorToken(name string) (bool, error) { func (d *DB) ListExecutorRegistrations() ([]ExecutorRegistration, error) { rows, err := d.Query(` - select e.name, t.created_at, t.expires_at, e.labels, - e.quarantine_reason, e.quarantined_at + select e.name, t.created_at, t.expires_at, e.labels from mill_executors e join mill_executor_tokens t on t.token_hash = e.token_hash order by e.name @@ -175,14 +146,12 @@ func (d *DB) ListExecutorRegistrations() ([]ExecutorRegistration, error) { var out []ExecutorRegistration for rows.Next() { var registration ExecutorRegistration - var expires, labelsRaw, quarantineReason, quarantinedAt sql.NullString + var expires, labelsRaw sql.NullString if err := rows.Scan( ®istration.Name, ®istration.CreatedAt, &expires, &labelsRaw, - &quarantineReason, - &quarantinedAt, ); err != nil { return nil, err } @@ -198,12 +167,6 @@ func (d *DB) ListExecutorRegistrations() ([]ExecutorRegistration, error) { return nil, err } } - if quarantineReason.Valid { - registration.QuarantineReason = &quarantineReason.String - } - if quarantinedAt.Valid { - registration.QuarantinedAt = &quarantinedAt.String - } out = append(out, registration) } return out, rows.Err() diff --git a/spindle/db/mill_tokens_test.go b/spindle/db/mill_tokens_test.go index 8cadb83b..8ba393bc 100644 --- a/spindle/db/mill_tokens_test.go +++ b/spindle/db/mill_tokens_test.go @@ -131,30 +131,6 @@ func TestRevokeExecutorToken(t *testing.T) { } } -func TestExecutorQuarantineIsVisibleAndReversible(t *testing.T) { - d := newTestDB(t) - createAndRegisterExecutor(t, d, "exec-1", "hash-1", nil, nil) - if err := d.QuarantineExecutor("exec-1", "missed cancel deadline"); err != nil { - t.Fatalf("QuarantineExecutor: %v", err) - } - if _, _, ok, err := d.ResolveExecutorToken("hash-1"); err != nil || ok { - t.Fatalf("ResolveExecutorToken(quarantined) = (ok=%v, err=%v), want (false, nil)", ok, err) - } - registrations, err := d.ListExecutorRegistrations() - if err != nil { - t.Fatalf("ListExecutorRegistrations: %v", err) - } - if len(registrations) != 1 || registrations[0].QuarantineReason == nil || *registrations[0].QuarantineReason != "missed cancel deadline" || registrations[0].QuarantinedAt == nil { - t.Fatalf("quarantined executor not surfaced: %+v", registrations) - } - if cleared, err := d.ClearExecutorQuarantine("exec-1"); err != nil || !cleared { - t.Fatalf("ClearExecutorQuarantine = (%v, %v), want (true, nil)", cleared, err) - } - if _, _, ok, err := d.ResolveExecutorToken("hash-1"); err != nil || !ok { - t.Fatalf("ResolveExecutorToken(cleared) = (ok=%v, err=%v), want (true, nil)", ok, err) - } -} - func TestListExecutorRegistrations(t *testing.T) { d := newTestDB(t) @@ -278,7 +254,7 @@ func TestLegacyExecutorTokensMigrateWithoutChangingIdentityState(t *testing.T) { ); insert into mill_executors ( name, token_hash, created_at, expires_at, labels, quarantine_reason, quarantined_at - ) values ('exec-1', 'hash-1', '2026-01-02T03:04:05Z', ?, '["arm64"]', null, null); + ) values ('exec-1', 'hash-1', '2026-01-02T03:04:05Z', ?, '["arm64"]', 'old quarantine', '2026-01-03T00:00:00Z'); insert into mill_executor_cursors (node_id, epoch, acked_seqno) values ('exec-1', 'epoch-1', 42); `, expires); err != nil { @@ -305,6 +281,15 @@ func TestLegacyExecutorTokensMigrateWithoutChangingIdentityState(t *testing.T) { if registrations[0].CreatedAt != "2026-01-02T03:04:05Z" || registrations[0].ExpiresAt == nil || registrations[0].ExpiresAt.Format(time.RFC3339) != expires { t.Fatalf("migrated credential metadata = %+v, want created and expiry preserved", registrations[0]) } + var quarantineReason, quarantinedAt sql.NullString + if err := d.QueryRow(` + select quarantine_reason, quarantined_at from mill_executors where name = 'exec-1' + `).Scan(&quarantineReason, &quarantinedAt); err != nil { + t.Fatalf("read migrated compatibility fields: %v", err) + } + if quarantineReason.Valid || quarantinedAt.Valid { + t.Fatalf("migrated compatibility fields were not cleared: %v, %v", quarantineReason, quarantinedAt) + } var acked uint64 if err := d.QueryRow(`select acked_seqno from mill_executor_cursors where node_id = 'exec-1' and epoch = 'epoch-1'`).Scan(&acked); err != nil { t.Fatalf("read migrated cursor: %v", err) @@ -313,3 +298,60 @@ func TestLegacyExecutorTokensMigrateWithoutChangingIdentityState(t *testing.T) { t.Fatalf("migrated cursor = %d, want 42", acked) } } + +func TestCurrentExecutorSchemaClearsLegacyQuarantineState(t *testing.T) { + path := filepath.Join(t.TempDir(), "spindle.db") + legacy, err := sql.Open("sqlite3", path) + if err != nil { + t.Fatalf("open legacy db: %v", err) + } + if _, err := legacy.Exec(` + create table mill_executor_tokens ( + token_hash text primary key, + created_at text not null, + expires_at text + ); + create table mill_executors ( + name text primary key, + token_hash text not null unique, + labels text, + quarantine_reason text, + quarantined_at text, + foreign key (token_hash) references mill_executor_tokens(token_hash) on delete cascade + ); + create table migrations ( + id integer primary key autoincrement, + name text unique + ); + insert into mill_executor_tokens (token_hash, created_at) + values ('hash-1', '2026-01-02T03:04:05Z'); + insert into mill_executors (name, token_hash, labels, quarantine_reason, quarantined_at) + values ('exec-1', 'hash-1', '["arm64"]', 'old quarantine', '2026-01-03T00:00:00Z'); + insert into migrations (name) values ('mill-executor-credentials'); + `); err != nil { + legacy.Close() + t.Fatalf("seed current db: %v", err) + } + if err := legacy.Close(); err != nil { + t.Fatalf("close current db: %v", err) + } + + d, err := Make(context.Background(), path) + if err != nil { + t.Fatalf("Make migrated db: %v", err) + } + defer d.Close() + name, labels, ok, err := d.ResolveExecutorToken("hash-1") + if err != nil || !ok || name != "exec-1" || !slices.Equal(labels, []string{"arm64"}) { + t.Fatalf("ResolveExecutorToken(migrated) = (%q, %v, %v, %v)", name, labels, ok, err) + } + var quarantineReason, quarantinedAt sql.NullString + if err := d.QueryRow(` + select quarantine_reason, quarantined_at from mill_executors where name = 'exec-1' + `).Scan(&quarantineReason, &quarantinedAt); err != nil { + t.Fatalf("read migrated compatibility fields: %v", err) + } + if quarantineReason.Valid || quarantinedAt.Valid { + t.Fatalf("migrated compatibility fields were not cleared: %v, %v", quarantineReason, quarantinedAt) + } +} diff --git a/spindle/mill/lease.go b/spindle/mill/lease.go index af8be700..2adc6ca9 100644 --- a/spindle/mill/lease.go +++ b/spindle/mill/lease.go @@ -161,6 +161,12 @@ func (l *RemoteLease) cancelRequested() (bool, string) { return l.cancel, l.reason } +func (l *RemoteLease) cancelPending() bool { + l.mu.Lock() + defer l.mu.Unlock() + return l.cancel && !l.cancelAcked && l.state != leaseDone +} + func (l *RemoteLease) releaseState() (leaseState, bool) { l.mu.Lock() defer l.mu.Unlock() diff --git a/spindle/mill/mill.go b/spindle/mill/mill.go index dfbab51c..ea5d36b3 100644 --- a/spindle/mill/mill.go +++ b/spindle/mill/mill.go @@ -36,12 +36,10 @@ const ( defaultBidTimeout = 5 * time.Second defaultTopK = 3 defaultMaxPending = 100 - defaultQuarantineStrikes = 3 defaultCancelAckTimeout = 10 * time.Second defaultCancelTeardownTimeout = 10 * time.Minute ) -// repeated protocol errors quarantine the node var errProtocolViolation = errors.New("executor protocol violation") func protoErrf(format string, args ...any) error { @@ -59,8 +57,6 @@ type Config struct { CancelAckTimeout time.Duration // teardown can take minutes after the cancel is acknowledged CancelTeardownTimeout time.Duration - // how many protocol-violating session deaths in a row quarantine the node - QuarantineStrikes int } type Mill struct { @@ -77,7 +73,6 @@ type Mill struct { leases map[string]*RemoteLease reservations map[string]*RemoteLease nodeSeqno map[string]uint64 - protoStrikes map[string]int quotaLeases map[string]*RemoteLease pending int changeCh chan struct{} @@ -107,9 +102,6 @@ func New(l *slog.Logger, cfg Config) *Mill { if cfg.CancelTeardownTimeout <= 0 { cfg.CancelTeardownTimeout = defaultCancelTeardownTimeout } - if cfg.QuarantineStrikes <= 0 { - cfg.QuarantineStrikes = defaultQuarantineStrikes - } return &Mill{ l: l, cfg: cfg, @@ -117,7 +109,6 @@ func New(l *slog.Logger, cfg Config) *Mill { leases: make(map[string]*RemoteLease), reservations: make(map[string]*RemoteLease), nodeSeqno: make(map[string]uint64), - protoStrikes: make(map[string]int), changeCh: make(chan struct{}), quotaLeases: make(map[string]*RemoteLease), } @@ -179,10 +170,19 @@ func (m *Mill) attachSession(sess *millSession) (uint64, bool) { if old.live(m.cfg.ReconnectGrace) { return 0, false } + if old.recoveryExpired { + m.l.Warn("rejecting executor reconnect while lease failure retries", "node", sess.nodeID) + return 0, false + } if old.graceTimer != nil { old.graceTimer.Stop() } old.close() + if old.recovering { + m.armSessionRecoveryLocked(sess, old.graceDeadline) + } else { + m.armSessionRecoveryLocked(sess, time.Now().Add(m.cfg.ReconnectGrace)) + } if old.disconnected { m.metrics.RecordMillReconnect() m.l.Info("executor reconnected", "node", sess.nodeID) @@ -210,6 +210,30 @@ func (m *Mill) touchSession(sess *millSession) bool { return true } +func (m *Mill) armSessionRecoveryLocked(sess *millSession, deadline time.Time) { + sess.recovering = true + sess.recoveryReady = false + sess.graceDeadline = deadline + delay := time.Until(deadline) + if delay < 0 { + delay = 0 + } + sess.graceTimer = time.AfterFunc(delay, func() { m.failLeasesAfterGrace(sess) }) +} + +func (m *Mill) maybeFinishSessionRecoveryLocked(sess *millSession) { + if m.sessions[sess.nodeID] != sess || sess.disconnected || !sess.recovering || sess.recoveryExpired || + !sess.recoveryReady || len(sess.pendingCancels) != 0 { + return + } + if sess.graceTimer != nil { + sess.graceTimer.Stop() + } + sess.recovering = false + sess.graceDeadline = time.Time{} + sess.graceTimer = nil +} + func (m *Mill) detachSession(sess *millSession) { m.mu.Lock() if m.sessions[sess.nodeID] != sess { @@ -219,7 +243,9 @@ func (m *Mill) detachSession(sess *millSession) { return } sess.disconnected = true - sess.graceTimer = time.AfterFunc(m.cfg.ReconnectGrace, func() { m.failLeasesAfterGrace(sess) }) + if !sess.recovering { + m.armSessionRecoveryLocked(sess, time.Now().Add(m.cfg.ReconnectGrace)) + } m.mu.Unlock() sess.close() @@ -227,31 +253,10 @@ func (m *Mill) detachSession(sess *millSession) { m.notifyChange() } -// repeated protocol errors quarantine the node until its stream is repaired func (m *Mill) noteSessionError(sess *millSession, err error) { - if !errors.Is(err, errProtocolViolation) { - m.mu.Lock() - delete(m.protoStrikes, sess.nodeID) - m.mu.Unlock() - return + if errors.Is(err, errProtocolViolation) { + m.l.Warn("executor session rejected on protocol violation", "node", sess.nodeID, "err", err) } - - m.mu.Lock() - m.protoStrikes[sess.nodeID]++ - strikes := m.protoStrikes[sess.nodeID] - m.mu.Unlock() - - m.l.Warn("executor session died on protocol violation", "node", sess.nodeID, "strikes", strikes, "err", err) - if strikes < m.cfg.QuarantineStrikes || m.db == nil { - return - } - - reason := fmt.Sprintf("repeated protocol violations (%d): %v", strikes, err) - if err := m.db.QuarantineExecutor(sess.nodeID, reason); err != nil { - m.l.Error("failed to quarantine executor", "node", sess.nodeID, "err", err) - return - } - m.l.Warn("executor quarantined; reset its stream state and unquarantine to readmit", "node", sess.nodeID) } func (m *Mill) sessionReady(sess *millSession) { @@ -281,22 +286,26 @@ func (m *Mill) cancelledLeasesForNode(nodeID string) []*RemoteLease { var leases []*RemoteLease for _, lease := range candidates { - if cancelled, _ := lease.cancelRequested(); cancelled && lease.getState() != leaseDone { + if lease.cancelPending() { leases = append(leases, lease) } } return leases } -// reconnect grace ran out without the executor coming back. fails every +// reconnect recovery did not complete before grace expired. fails every // lease the node held, and reschedules itself if any failure didn't stick func (m *Mill) failLeasesAfterGrace(sess *millSession) { + m.failLeasesAfterGraceAttempt(sess, false) +} + +func (m *Mill) failLeasesAfterGraceAttempt(sess *millSession, retry bool) { m.mu.Lock() - if m.sessions[sess.nodeID] != sess || !sess.disconnected { - // reconnected in the meantime + if m.sessions[sess.nodeID] != sess || !sess.recovering || (sess.recoveryExpired && !retry) { m.mu.Unlock() return } + sess.recoveryExpired = true var dead []*RemoteLease for _, lease := range m.leases { if lease.nodeID == sess.nodeID { @@ -337,16 +346,20 @@ func (m *Mill) failLeasesAfterGrace(sess *millSession) { if !success { // something didn't finish cleanly, try again shortly. finished // leases skip themselves on the next pass - sess.graceTimer = time.AfterFunc(5*time.Second, func() { m.failLeasesAfterGrace(sess) }) + sess.graceTimer = time.AfterFunc(5*time.Second, func() { m.failLeasesAfterGraceAttempt(sess, true) }) m.mu.Unlock() return } // everything failed cleanly. forget the node and wake placement - if m.sessions[sess.nodeID] == sess { + current := m.sessions[sess.nodeID] == sess + if current { delete(m.sessions, sess.nodeID) } m.mu.Unlock() + if current { + sess.close() + } m.notifyChange() m.sweepUnclaimedOrphans() } @@ -800,7 +813,7 @@ func (m *Mill) rankCandidates(engineName string, requiredLabels []string) []*mil var rs []ranked for _, s := range m.sessions { // only live, reporting sessions can take work - if s.disconnected { + if s.disconnected || s.recovering || len(s.pendingCancels) != 0 { continue } if s.snapshot == nil { @@ -1173,13 +1186,59 @@ func (m *Mill) releaseRemote(lease *RemoteLease) { } func (m *Mill) sendCancel(sess *millSession, lease *RemoteLease, reason string) { + lease.finishMu.Lock() + m.mu.Lock() + if m.sessions[lease.nodeID] != sess || sess.disconnected || m.leases[lease.id] != lease || lease.getState() == leaseDone { + m.mu.Unlock() + lease.finishMu.Unlock() + return + } + sess.pendingCancels[lease.id] = struct{}{} + m.mu.Unlock() + lease.finishMu.Unlock() + if err := sess.send(&millproto.Message{CancelAttempt: &millv1.CancelAttempt{ LeaseId: lease.id, Reason: reason, }}); err != nil { + m.settleSessionCancel(sess, lease.id) + sess.close() + return + } + time.AfterFunc(m.cfg.CancelAckTimeout, func() { m.checkCancelAck(sess, lease) }) +} + +func (m *Mill) sendUntrackedCancel(sess *millSession, leaseID, reason string) error { + m.mu.Lock() + if m.sessions[sess.nodeID] != sess || sess.disconnected { + m.mu.Unlock() + return errSessionClosed + } + sess.pendingCancels[leaseID] = struct{}{} + m.mu.Unlock() + + if err := sess.send(&millproto.Message{CancelAttempt: &millv1.CancelAttempt{ + LeaseId: leaseID, + Reason: reason, + }}); err != nil { + m.settleSessionCancel(sess, leaseID) + sess.close() + return err + } + time.AfterFunc(m.cfg.CancelAckTimeout, func() { m.checkUntrackedCancelAck(sess, leaseID) }) + return nil +} + +func (m *Mill) checkUntrackedCancelAck(sess *millSession, leaseID string) { + m.mu.Lock() + _, pending := sess.pendingCancels[leaseID] + isCurrent := m.sessions[sess.nodeID] == sess && !sess.disconnected + m.mu.Unlock() + if !pending || !isCurrent { return } - time.AfterFunc(m.cfg.CancelAckTimeout, func() { m.checkCancelAck(lease) }) + m.l.Warn("node did not acknowledge unknown lease cancel within deadline; closing session", "node", sess.nodeID, "lease", leaseID, "deadline", m.cfg.CancelAckTimeout) + sess.close() } func (m *Mill) sessionForNode(nodeID string) *millSession { @@ -1210,6 +1269,8 @@ func (m *Mill) onSnapshot(sess *millSession, snap *millv1.NodeSnapshot) error { lease.claimed = true } } + sess.recoveryReady = true + m.maybeFinishSessionRecoveryLocked(sess) m.mu.Unlock() m.notifyChange() return nil @@ -1254,6 +1315,7 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { ar *millv1.AttemptResult } var pendingTerminals []pendingTerminal + var terminalCancelIDs []string var artifactLeases []*RemoteLease finishedInBatch := make(map[string]struct{}) type startupObservation struct { @@ -1294,6 +1356,9 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { applyFunc := func(tx *db.EventBatchTx) error { for _, entry := range newEntries { + if entry.GetAttemptResult() != nil { + terminalCancelIDs = append(terminalCancelIDs, entry.LeaseId) + } m.mu.Lock() lease := m.leases[entry.LeaseId] m.mu.Unlock() @@ -1462,6 +1527,9 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { } m.mu.Unlock() + for _, leaseID := range terminalCancelIDs { + m.settleSessionCancel(sess, leaseID) + } for _, pt := range pendingTerminals { if pt.lease.millRecordsTerminalMetrics && pt.ar.GetMillRecordsTerminalMetrics() && @@ -1566,9 +1634,22 @@ func (m *Mill) onLiveLog(sess *millSession, ll *millv1.LiveLog) error { return nil } +func (m *Mill) settleSessionCancel(sess *millSession, leaseID string) { + m.mu.Lock() + _, pending := sess.pendingCancels[leaseID] + delete(sess.pendingCancels, leaseID) + m.maybeFinishSessionRecoveryLocked(sess) + m.mu.Unlock() + if pending { + m.notifyChange() + } +} + func (m *Mill) onCancelAck(sess *millSession, ca *millv1.CancelAck) { + leaseID := ca.GetLeaseId() + m.settleSessionCancel(sess, leaseID) m.mu.Lock() - lease := m.leases[ca.GetLeaseId()] + lease := m.leases[leaseID] m.mu.Unlock() if lease == nil || lease.nodeID != sess.nodeID { return @@ -1579,34 +1660,25 @@ func (m *Mill) onCancelAck(sess *millSession, ca *millv1.CancelAck) { time.AfterFunc(m.cfg.CancelTeardownTimeout, func() { m.checkCancelTeardown(lease) }) } -func (m *Mill) checkCancelAck(lease *RemoteLease) { +func (m *Mill) checkCancelAck(sess *millSession, lease *RemoteLease) { lease.mu.Lock() isDone := (lease.state == leaseDone) isAcked := lease.cancelAcked lease.mu.Unlock() if isDone || isAcked { + m.settleSessionCancel(sess, lease.id) return } m.mu.Lock() - sess := m.sessions[lease.nodeID] - disconnected := sess == nil || sess.disconnected + isCurrent := m.sessions[lease.nodeID] == sess && !sess.disconnected m.mu.Unlock() - if disconnected { - // reconnect replays the cancel, the grace timer handles nodes that never return + if !isCurrent { return } - m.l.Warn("node did not acknowledge cancel request within deadline, quarantining", "node", lease.nodeID, "lease", lease.id, "deadline", m.cfg.CancelAckTimeout) - - reason := fmt.Sprintf("cancel unacknowledged for lease %s after %s", lease.id, m.cfg.CancelAckTimeout) - if m.db != nil { - if err := m.db.QuarantineExecutor(lease.nodeID, reason); err != nil { - m.l.Error("failed to quarantine executor", "node", lease.nodeID, "err", err) - } - } - + m.l.Warn("node did not acknowledge cancel request within deadline; closing session", "node", lease.nodeID, "lease", lease.id, "deadline", m.cfg.CancelAckTimeout) sess.close() } diff --git a/spindle/mill/mill_test.go b/spindle/mill/mill_test.go index 04b8c320..4c1501e1 100644 --- a/spindle/mill/mill_test.go +++ b/spindle/mill/mill_test.go @@ -489,6 +489,46 @@ func TestAttachSessionReplacesSilentIncumbentButRejectsActiveDuplicate(t *testin } } +func TestSilentSessionReplacementWithoutSnapshotFailsLeaseAtDeadline(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute}) + old := newSession("node-1", "inc-old", nil, nopEncoder(), discardLogger()) + if _, ok := m.attachSession(old); !ok { + t.Fatal("first attach rejected") + } + lease := newLease("lease-1", old.nodeID, old.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + if err := m.persistLease(lease, leaseRowRunning); err != nil { + t.Fatalf("persistLease: %v", err) + } + m.mu.Lock() + m.leases[lease.id] = lease + old.lastSeen = time.Now().Add(-2 * m.cfg.ReconnectGrace) + m.mu.Unlock() + + replacement := newSession("node-1", "inc-new", nil, nopEncoder(), discardLogger()) + attachStarted := time.Now() + if _, ok := m.attachSession(replacement); !ok { + t.Fatal("silent incumbent blocked replacement") + } + attachFinished := time.Now() + if !replacement.recovering || replacement.graceTimer == nil || replacement.graceDeadline.IsZero() { + t.Fatal("replacement without snapshot has no recovery deadline") + } + if replacement.graceDeadline.Before(attachStarted.Add(m.cfg.ReconnectGrace)) || + replacement.graceDeadline.After(attachFinished.Add(m.cfg.ReconnectGrace)) { + t.Fatalf("replacement deadline = %v, want one reconnect grace after attach", replacement.graceDeadline) + } + m.failLeasesAfterGrace(replacement) + + m.mu.Lock() + _, stillActive := m.leases[lease.id] + m.mu.Unlock() + if stillActive { + t.Fatal("replacement without snapshot stranded incumbent lease") + } +} + func TestPlaceReleasesRemoteReservationWhenInitialPersistenceFails(t *testing.T) { m, bdb := restoreTestMill(t, Config{BidTimeout: time.Second}) if err := bdb.Close(); err != nil { @@ -767,7 +807,7 @@ func TestReplacementLostBeforeSnapshotFailsOldEpochLease(t *testing.T) { if _, ok := m.attachSession(replacement); !ok { t.Fatal("attachSession rejected replacement") } - replacement.disconnected = true + m.detachSession(replacement) m.failLeasesAfterGrace(replacement) m.mu.Lock() @@ -831,7 +871,7 @@ func TestClaimedSweep(t *testing.T) { } } -func TestUnacknowledgedCancelQuarantinesNode(t *testing.T) { +func TestUnacknowledgedCancelClosesSession(t *testing.T) { m, bdb := restoreTestMill(t, Config{ReconnectGrace: time.Minute, CancelAckTimeout: 50 * time.Millisecond}) registerTestExecutor(t, bdb, "node-1", HashToken("tok-1"), nil) @@ -857,8 +897,212 @@ func TestUnacknowledgedCancelQuarantinesNode(t *testing.T) { case <-time.After(2 * time.Second): t.Fatal("session was not closed after the cancel went unacknowledged") } - if _, _, ok, err := bdb.ResolveExecutorToken(HashToken("tok-1")); err != nil || ok { - t.Fatalf("executor still admitted after ignoring a cancel: ok = %v, err = %v", ok, err) + if _, _, ok, err := bdb.ResolveExecutorToken(HashToken("tok-1")); err != nil || !ok { + t.Fatalf("executor identity rejected after session timeout: ok = %v, err = %v", ok, err) + } +} + +func TestPendingCancelBlocksPlacementUntilAcknowledged(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute, CancelAckTimeout: time.Minute}) + sess := addCandidateSession(t, m, "node-1", nil, 0, nil) + lease := newLease("lease-1", "node-1", sess.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + + assertRankedNodes(t, m.rankCandidates("dummy", nil), []string{"node-1"}) + m.destroy(lease.wid) + assertRankedNodes(t, m.rankCandidates("dummy", nil), nil) + + m.onCancelAck(sess, &millv1.CancelAck{LeaseId: lease.id}) + assertRankedNodes(t, m.rankCandidates("dummy", nil), []string{"node-1"}) +} + +func TestTerminalSettlesPendingCancelPlacementGate(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute, CancelAckTimeout: time.Minute}) + sess := addCandidateSession(t, m, "node-1", nil, 0, nil) + lease := newLease("lease-1", sess.nodeID, sess.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + + m.destroy(lease.wid) + assertRankedNodes(t, m.rankCandidates("dummy", nil), nil) + if err := m.onEventBatch(sess, &millv1.EventBatch{ + Epoch: sess.epoch, + Events: []*millv1.Event{{ + Seqno: 1, + LeaseId: lease.id, + Payload: &millv1.Event_AttemptResult{AttemptResult: &millv1.AttemptResult{ + Status: millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED, + }}, + }}, + }); err != nil { + t.Fatalf("onEventBatch: %v", err) + } + assertRankedNodes(t, m.rankCandidates("dummy", nil), []string{"node-1"}) +} + +func TestCancelSentAfterTerminalDoesNotLeakPlacementGate(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute, CancelAckTimeout: time.Minute}) + sess := addCandidateSession(t, m, "node-1", nil, 0, nil) + lease := newLease("lease-1", sess.nodeID, sess.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + if err := m.persistLease(lease, leaseRowRunning); err != nil { + t.Fatalf("persistLease: %v", err) + } + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + + lease.requestCancel("done") + if err := m.finishLiveLease(lease, string(models.StatusKindCancelled), "done"); err != nil { + t.Fatalf("finishLiveLease: %v", err) + } + m.sendCancel(sess, lease, "done") + + m.mu.Lock() + pending := len(sess.pendingCancels) + m.mu.Unlock() + if pending != 0 { + t.Fatalf("pending cancels after terminal = %d, want 0", pending) + } + assertRankedNodes(t, m.rankCandidates("dummy", nil), []string{"node-1"}) +} + +func TestValidSnapshotRecoversAfterProtocolError(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute}) + first := addCandidateSession(t, m, "node-1", nil, 0, nil) + lease := newLease("lease-1", first.nodeID, first.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + + m.noteSessionError(first, protoErrf("bad stream")) + m.detachSession(first) + replacement := newSession(first.nodeID, first.epoch, nil, nopEncoder(), discardLogger()) + if _, ok := m.attachSession(replacement); !ok { + t.Fatal("reconnect was rejected") + } + m.sessionReady(replacement) + if err := m.onSnapshot(replacement, &millv1.NodeSnapshot{ + Seqno: 1, + ActiveLeaseIds: []string{lease.id}, + Engines: map[string]*millv1.EngineAvailability{ + "dummy": {Available: true}, + }, + }); err != nil { + t.Fatalf("onSnapshot: %v", err) + } + + m.mu.Lock() + recovering := replacement.recovering + m.mu.Unlock() + if recovering { + t.Fatal("valid reconnect snapshot did not end recovery") + } + assertRankedNodes(t, m.rankCandidates("dummy", nil), []string{"node-1"}) +} + +func TestReconnectRetriesKeepOriginalFailureDeadline(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute}) + first := addCandidateSession(t, m, "node-1", nil, 0, nil) + lease := newLease("lease-1", first.nodeID, first.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + if err := m.persistLease(lease, leaseRowRunning); err != nil { + t.Fatalf("persistLease: %v", err) + } + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + + m.detachSession(first) + deadline := first.graceDeadline + second := newSession(first.nodeID, first.epoch, nil, nopEncoder(), discardLogger()) + if _, ok := m.attachSession(second); !ok { + t.Fatal("first reconnect was rejected") + } + m.detachSession(second) + third := newSession(first.nodeID, first.epoch, nil, nopEncoder(), discardLogger()) + if _, ok := m.attachSession(third); !ok { + t.Fatal("second reconnect was rejected") + } + if !third.graceDeadline.Equal(deadline) { + t.Fatalf("reconnect moved failure deadline from %v to %v", deadline, third.graceDeadline) + } + + m.failLeasesAfterGrace(third) + m.mu.Lock() + _, tracked := m.leases[lease.id] + m.mu.Unlock() + if tracked { + t.Fatal("reconnect retries kept the lease past its original failure deadline") + } +} + +func TestUnacknowledgedCancelReconnectsReachBoundedTerminal(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute, CancelAckTimeout: time.Minute}) + first := addCandidateSession(t, m, "node-1", nil, 0, nil) + lease := newLease("lease-1", first.nodeID, first.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + if err := m.persistLease(lease, leaseRowRunning); err != nil { + t.Fatalf("persistLease: %v", err) + } + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + + m.destroy(lease.wid) + m.checkCancelAck(first, lease) + m.detachSession(first) + deadline := first.graceDeadline + + second := newSession(first.nodeID, first.epoch, nil, nopEncoder(), discardLogger()) + if _, ok := m.attachSession(second); !ok { + t.Fatal("first reconnect was rejected") + } + m.sessionReady(second) + if err := m.onSnapshot(second, &millv1.NodeSnapshot{ + Seqno: 1, + ActiveLeaseIds: []string{lease.id}, + Engines: map[string]*millv1.EngineAvailability{ + "dummy": {Available: true}, + }, + }); err != nil { + t.Fatalf("onSnapshot: %v", err) + } + if !second.recovering { + t.Fatal("snapshot cleared recovery while cancel was still pending") + } + m.checkCancelAck(second, lease) + m.detachSession(second) + + third := newSession(first.nodeID, first.epoch, nil, nopEncoder(), discardLogger()) + if _, ok := m.attachSession(third); !ok { + t.Fatal("second reconnect was rejected") + } + m.sessionReady(third) + if !third.graceDeadline.Equal(deadline) { + t.Fatalf("cancel reconnect moved failure deadline from %v to %v", deadline, third.graceDeadline) + } + m.failLeasesAfterGrace(third) + + select { + case result := <-lease.terminal: + if result.GetStatus() != millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED { + t.Fatalf("terminal status = %v, want CANCELLED", result.GetStatus()) + } + case <-time.After(time.Second): + t.Fatal("unacknowledged cancel did not reach a bounded terminal") } } @@ -913,31 +1157,92 @@ func TestAckedCancelFinishesOnlyItsLeaseAfterTeardownDeadline(t *testing.T) { default: } if _, _, ok, err := bdb.ResolveExecutorToken(HashToken("tok-1")); err != nil || !ok { - t.Fatalf("node quarantined for a slow teardown: ok = %v, err = %v", ok, err) + t.Fatalf("executor identity rejected for a slow teardown: ok = %v, err = %v", ok, err) } } -func TestCancelAckDeadlineSparesDisconnectedNode(t *testing.T) { +func TestLateCancelAckTimerKeepsReconnectGrace(t *testing.T) { m, bdb := restoreTestMill(t, Config{ReconnectGrace: time.Minute, CancelAckTimeout: time.Minute}) registerTestExecutor(t, bdb, "node-1", HashToken("tok-1"), nil) sess := newSession("node-1", "inc-1", nil, nopEncoder(), discardLogger()) m.attachSession(sess) + if err := m.onSnapshot(sess, &millv1.NodeSnapshot{Seqno: 1}); err != nil { + t.Fatalf("onSnapshot: %v", err) + } lease := newLease("lease-1", "node-1", "inc-1", "dummy") lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} lease.setState(leaseRunning) + if err := m.persistLease(lease, leaseRowRunning); err != nil { + t.Fatalf("persistLease: %v", err) + } m.mu.Lock() m.leases[lease.id] = lease m.mu.Unlock() m.destroy(lease.wid) + m.onCancelAck(sess, &millv1.CancelAck{LeaseId: lease.id}) m.detachSession(sess) + m.checkCancelAck(sess, lease) - m.checkCancelAck(lease) - + m.mu.Lock() + recovering := sess.recovering + timerArmed := sess.graceTimer != nil + m.mu.Unlock() + if !recovering || !timerArmed { + t.Fatal("late cancel timer disarmed reconnect grace") + } + m.failLeasesAfterGrace(sess) + select { + case result := <-lease.terminal: + if result.GetStatus() != millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED { + t.Fatalf("terminal status = %v, want CANCELLED", result.GetStatus()) + } + case <-time.After(time.Second): + t.Fatal("reconnect grace did not finish the disconnected lease") + } if _, _, ok, err := bdb.ResolveExecutorToken(HashToken("tok-1")); err != nil || !ok { - t.Fatalf("node quarantined for a cancel it could not answer: ok = %v, err = %v", ok, err) + t.Fatalf("executor identity rejected for a cancel it could not answer: ok = %v, err = %v", ok, err) + } +} + +func TestCancelAckDeadlineDoesNotCloseReplacementSession(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute, CancelAckTimeout: time.Minute}) + oldSession := newSession("node-1", "inc-1", nil, nopEncoder(), discardLogger()) + m.attachSession(oldSession) + lease := newLease("lease-1", "node-1", oldSession.epoch, "dummy") + lease.wid = models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} + lease.setState(leaseRunning) + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + m.destroy(lease.wid) + m.detachSession(oldSession) + + replacementClosed := make(chan struct{}) + replacement := newSession("node-1", "inc-1", nil, nopEncoder(), discardLogger()) + replacement.closeTransport = func() error { + close(replacementClosed) + return nil + } + if _, ok := m.attachSession(replacement); !ok { + t.Fatal("replacement session was rejected") + } + replacement.snapshot = &millv1.NodeSnapshot{ + Seqno: 1, + Engines: map[string]*millv1.EngineAvailability{ + "dummy": {Available: true}, + }, + } + m.sessionReady(replacement) + assertRankedNodes(t, m.rankCandidates("dummy", nil), nil) + + m.checkCancelAck(oldSession, lease) + select { + case <-replacementClosed: + t.Fatal("cancel timer from old session closed its replacement") + default: } } @@ -1039,38 +1344,16 @@ func TestBoundedBidding(t *testing.T) { } } -func TestProtocolStrikesQuarantineExecutor(t *testing.T) { - m, bdb := restoreTestMill(t, Config{ReconnectGrace: time.Minute, QuarantineStrikes: 2}) +func TestProtocolErrorsDoNotRejectExecutorIdentity(t *testing.T) { + m, bdb := restoreTestMill(t, Config{ReconnectGrace: time.Minute}) registerTestExecutor(t, bdb, "node-1", HashToken("tok-1"), nil) sess := newSession("node-1", "inc-1", nil, nopEncoder(), discardLogger()) - resolve := func() bool { - _, _, ok, err := bdb.ResolveExecutorToken(HashToken("tok-1")) - if err != nil { - t.Fatalf("ResolveExecutorToken: %v", err) - } - return ok - } - - m.noteSessionError(sess, protoErrf("gap in stream seqnos. Expected 1, got 9")) - if !resolve() { - t.Fatal("quarantined on first strike, want threshold 2") - } - - // a session that ends cleanly resets the count - m.noteSessionError(sess, io.EOF) - m.noteSessionError(sess, protoErrf("gap in stream seqnos. Expected 1, got 9")) - if !resolve() { - t.Fatal("transport death failed to reset the strike count") + for range 3 { + m.noteSessionError(sess, protoErrf("gap in stream seqnos. Expected 1, got 9")) } - - m.noteSessionError(sess, protoErrf("gap in stream seqnos. Expected 1, got 9")) - if resolve() { - t.Fatal("expected quarantine after consecutive protocol deaths") - } - registrations, err := bdb.ListExecutorRegistrations() - if err != nil || len(registrations) != 1 || registrations[0].QuarantineReason == nil { - t.Fatalf("quarantine not visible in list: %+v, %v", registrations, err) + if _, _, ok, err := bdb.ResolveExecutorToken(HashToken("tok-1")); err != nil || !ok { + t.Fatalf("executor identity rejected after protocol errors: ok = %v, err = %v", ok, err) } } diff --git a/spindle/mill/restore.go b/spindle/mill/restore.go index ee86e776..d29892cd 100644 --- a/spindle/mill/restore.go +++ b/spindle/mill/restore.go @@ -10,7 +10,6 @@ import ( "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" - millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" ) @@ -191,7 +190,7 @@ func (m *Mill) reconcileLeases(sess *millSession, activeLeaseIDs []string) error } } for _, id := range unknown { - if err := sess.send(&millproto.Message{CancelAttempt: &millv1.CancelAttempt{LeaseId: id, Reason: "lease is not owned by this mill"}}); err != nil { + if err := m.sendUntrackedCancel(sess, id, "lease is not owned by this mill"); err != nil { return fmt.Errorf("cancel unknown executor lease %q: %w", id, err) } } diff --git a/spindle/mill/restore_test.go b/spindle/mill/restore_test.go index 8b7ecb6c..d0b3f301 100644 --- a/spindle/mill/restore_test.go +++ b/spindle/mill/restore_test.go @@ -219,6 +219,9 @@ func TestSnapshotReconciliationCancelsUnknownExecutorLease(t *testing.T) { if err := m.onSnapshot(sess, &millv1.NodeSnapshot{ Seqno: 1, ActiveLeaseIds: []string{"executor-only"}, + Engines: map[string]*millv1.EngineAvailability{ + "dummy": {Available: true}, + }, }); err != nil { t.Fatalf("onSnapshot: %v", err) } @@ -230,6 +233,20 @@ func TestSnapshotReconciliationCancelsUnknownExecutorLease(t *testing.T) { default: t.Fatal("snapshot reconciliation left an executor-only lease running") } + assertRankedNodes(t, m.rankCandidates("dummy", nil), nil) + if err := m.onEventBatch(sess, &millv1.EventBatch{ + Epoch: sess.epoch, + Events: []*millv1.Event{{ + Seqno: 1, + LeaseId: "executor-only", + Payload: &millv1.Event_AttemptResult{AttemptResult: &millv1.AttemptResult{ + Status: millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED, + }}, + }}, + }); err != nil { + t.Fatalf("onEventBatch: %v", err) + } + assertRankedNodes(t, m.rankCandidates("dummy", nil), []string{"node-1"}) } func TestSweepFailsOrphansOfAbsentExecutors(t *testing.T) { diff --git a/spindle/mill/session.go b/spindle/mill/session.go index fc99752b..08851420 100644 --- a/spindle/mill/session.go +++ b/spindle/mill/session.go @@ -24,12 +24,16 @@ type millSession struct { l *slog.Logger closeTransport func() error - // snapshot, disconnected and graceTimer are guarded by Mill.mu, the - // fleet ranks across sessions under its own lock - snapshot *millv1.NodeSnapshot - disconnected bool - graceTimer *time.Timer - lastSeen time.Time + // recovery and cancellation state is guarded by Mill.mu + snapshot *millv1.NodeSnapshot + disconnected bool + recovering bool + recoveryReady bool + recoveryExpired bool + graceDeadline time.Time + graceTimer *time.Timer + lastSeen time.Time + pendingCancels map[string]struct{} mu sync.Mutex pending map[string]chan *millproto.Message // maps lease ID to response waiter @@ -44,14 +48,15 @@ type messageEncoder interface { func newSession(nodeID string, epoch string, labels []string, enc messageEncoder, l *slog.Logger) *millSession { return &millSession{ - nodeID: nodeID, - epoch: epoch, - labels: labels, - enc: enc, - l: l, - pending: make(map[string]chan *millproto.Message), - closed: make(chan struct{}), - lastSeen: time.Now(), + nodeID: nodeID, + epoch: epoch, + labels: labels, + enc: enc, + l: l, + pending: make(map[string]chan *millproto.Message), + pendingCancels: make(map[string]struct{}), + closed: make(chan struct{}), + lastSeen: time.Now(), } } -- 2.54.0