fix(meta): stale in-memory worker entries can survive an SQL-election leader handoff

[zwang28]

Summary

With multiple meta replicas using the SQL election backend, an outgoing leader can finish an expired-worker cleanup iteration after leadership has moved to another meta replica.

If the incoming leader initializes its in-memory worker snapshot before that deletion completes, the deletion is only reflected in the shared SQL metadata and the outgoing meta's local notification channel. The incoming leader does not receive the deletion and retains the old worker ID in memory.

When the compute node re-registers at the same host:port, it receives a new worker ID. The incoming leader can then hold both the old and new IDs for the same physical compute endpoint.

Impact

The duplicated worker identities can prevent barrier recovery indefinitely:

1. ControlStreamManager creates one control stream per worker ID.
2. Both IDs resolve to the same compute endpoint.
3. The compute barrier worker accepts only one control stream and resets the previous stream when another arrives.
4. Meta marks the disconnected identity as reconnecting and opens another stream.
5. The two identities continuously replace each other.
6. Initial barrier injection fails whenever one of the required identities is reconnecting:

failed to inject initial barrier ... unconnected worker node ...

This can leave databases stuck in recovery even though the physical compute process is reachable.

The inconsistent SQL and in-memory worker state can also trigger a panic in list_workers, where worker information loaded from SQL is combined with worker_extra_info using unwrap().

Relevant code

Expired-worker cleanup is not fenced by the leader term

In src/meta/src/controller/cluster.rs, the heartbeat checker:

1. Collects expired worker IDs.
2. Loads their host:port values from SQL.
3. Drops the controller lock.
4. Deletes each worker by host:port.

There is no leadership or term check immediately before the deletion:

let worker_to_delete = inner
    .worker_extra_info
    .iter()
    .filter(|(_, info)| info.expire_at.unwrap() < now)
    .map(|(id, _)| *id)
    .collect_vec();

let worker_infos = Worker::find()
    // ...
    .filter(worker::Column::WorkerId.is_in(worker_to_delete.clone()))
    .all(&inner.db)
    .await?;

drop(inner);

for (worker_id, worker_type, host, port) in worker_infos {
    let host_addr = PbHostAddress { host, port };
    cluster_controller.delete_worker(host_addr.clone()).await?;
}

A heartbeat-check iteration that started under the old leader can therefore mutate shared metadata after a new leader has initialized.

Deletion uses host:port instead of the collected worker ID

ClusterControllerInner::delete_worker resolves the row again using host:port:

let worker = Worker::find()
    .filter(
        worker::Column::Host
            .eq(host_addr.host)
            .and(worker::Column::Port.eq(host_addr.port)),
    )
    .one(&self.db)
    .await?;

Although the expiration loop already has the immutable worker_id, that ID is not passed to the deletion operation. This creates an additional identity race if the endpoint is re-registered.

Active worker state depends on process-local notifications

In src/meta/src/manager/metadata.rs, ActiveStreamingWorkerNodes initializes a snapshot and subsequently applies LocalNotification events:

LocalNotification::WorkerNodeDeleted(worker) => {
    self.worker_nodes.remove(&worker.id);
}

LocalNotification::WorkerNodeActivated(worker) => {
    self.worker_nodes.insert(worker.id, worker);
}

Notifications are local to a meta process. A deletion performed by the outgoing leader does not update a snapshot already initialized on the incoming leader.

The map is keyed only by worker ID, so activating a new ID for the same host:port does not replace the stale entry.

Duplicate IDs create a control-stream loop

In src/stream/src/task/barrier_worker/mod.rs, a new control stream resets the existing stream:

LocalActorOperation::NewControlStream { handle, init_request } => {
    self.control_stream_handle.reset_stream_with_err(
        Status::internal("control stream has been reset to a new one")
    );
    // ...
}

In src/meta/src/barrier/rpc.rs, meta reconnects a worker whenever its response stream is reset. Barrier injection rejects workers currently in the reconnecting state:

let node = if let Some((_, WorkerNodeState::Connected {
    control_stream, ..
})) = self.workers.get(worker_id)
{
    control_stream
} else {
    return Err(anyhow!("unconnected worker node {}", worker_id).into());
};

Suspected race

A possible interleaving is:

1. Meta A is leader and has worker W-old for endpoint H.
2. Meta A's heartbeat checker determines that W-old has expired.
3. SQL election or meta-store connectivity is delayed.
4. Meta A loses leadership while its cleanup iteration is still running.
5. Meta B becomes leader and initializes its active-worker snapshot containing W-old.
6. Meta A deletes W-old from SQL and emits only a process-local deletion notification.
7. The compute node at H re-registers through meta B and receives W-new.
8. Meta B adds W-new but never removes W-old.
9. Barrier recovery opens control streams for both IDs at endpoint H, causing an endless reset/reconnect loop.

Proposed fixes

The primary fix should make worker mutations safe across leader handoffs:

• Fence expired-worker cleanup using the current election term or leadership token.
• Recheck leadership immediately before committing each leader-only metadata mutation.
• Delete expired workers by immutable worker_id, not by host:port.
• Make the SQL deletion conditional on the expected worker identity.

The incoming leader should also be able to recover from missed local notifications:

• Reconcile ActiveStreamingWorkerNodes with persistent SQL metadata after leadership is acquired.
• Periodically reconcile the in-memory snapshot in release builds.
• Remove entries that no longer exist in persistent metadata.
• Detect multiple worker IDs resolving to the same host:port.

Useful defensive measures:

• When activating a worker, replace or reject any stale entry with the same endpoint.
• Deduplicate endpoints before ControlStreamManager opens control streams.
• Replace worker_extra_info lookups using unwrap() with an error or snapshot reload.

Test plan

Add a deterministic SQL-backend HA test using failpoints:

1. Start two meta replicas.
2. Register one streaming compute worker.
3. Pause the old leader's heartbeat checker after it loads an expired worker but before deletion.
4. Force a leader handoff.
5. Allow the new leader to initialize its worker snapshot.
6. Resume the old leader's deletion.
7. Re-register the compute endpoint.
8. Verify that the new leader contains only the current worker ID.
9. Verify that only one control stream is opened to the endpoint.
10. Verify that initial barrier recovery completes.

Acceptance criteria

• A meta replica cannot delete or mutate workers after losing leadership.
• Expired workers are deleted using their worker ID.
• A newly elected leader converges its in-memory worker state with SQL metadata.
• A physical host:port cannot remain represented by multiple active worker IDs.
• The race does not cause repeated control-stream resets or block barrier recovery.
• Missing worker_extra_info cannot panic list_workers.

Temporary workaround

Run with one meta replica until the leader-handoff race is fixed. This avoids cross-meta stale snapshots but temporarily removes automatic meta failover.