fix: keep load agents connected before start

2026-09-28 00:16:51 +08:00
parent 6b8ed4a2cc
commit 622fc50418
8 changed files with 352 additions and 58 deletions
@@ -146,9 +146,10 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}"
 1. 机器 B 建立 1000 条 WebSocket 连接并订阅对应 P2P Topic 和群 Topic。
 2. agent 向机器 A coordinator 报告 READY。
 3. 如果机器 A 尚未 READY，当前命令会等待，这是正常现象。
-4. coordinator 下发 `run_id`、场景和统一 UTC 开始时间。
-5. agent 完成发送和 WS 到达统计，等待 `history_settle`（默认 5 分钟）后，再以有界并发执行历史持久化回查；等待期间不要停止进程。
-6. agent 写出 summary、上报 coordinator，然后正常退出。
+4. 等待期间连接断开会自动重连并重新订阅；在 A/B 都实时保持 1000 条连接并完成最终屏障前，agent 不会退出或开始发压。
+5. coordinator 下发 `run_id`、场景和统一 UTC 开始时间。
+6. agent 完成发送和 WS 到达统计，等待 `history_settle`（默认 5 分钟）后，再以有界并发执行历史持久化回查；等待期间不要停止进程。
+7. agent 写出 summary、上报 coordinator，然后正常退出。

 机器 B 不需要传场景参数。`p2p`、`group` 或 `mixed` 由机器 A coordinator 决定。

@@ -42,6 +42,8 @@
 - `8090` 是无鉴权的压测协调端口，只允许压测内网访问，不要暴露到公网。
 - 两台机器必须使用 NTP/chrony 同步时间。报告会分别输出同机时钟延迟和跨机器单向延迟；跨机器出现负延迟样本时，后者仅用于诊断，不参与门禁。
 - agent 默认每 20 秒发送 WebSocket Ping，覆盖等待另一台机器、发压、drain 和历史核验阶段；`ws_ping_interval` 必须小于公网入口的空闲连接超时。
+- 正式开始前 agent 会持续检查实际存活连接；连接断开后自动重新登录并订阅 P2P、群 Topic。初次连接不齐或等待另一台机器期间掉线不会退出。
+- coordinator 使用两阶段屏障：两边实时连接数都达到配置值后进入验证倒计时；倒计时内任一方掉线就撤销屏障。两边再次确认满连接后才提交最终开始时间。

 ## 统计口径

@@ -24,6 +24,7 @@ type agentRun struct {
 	runID          string
 	scenario       string
 	clients        map[int]*WSClient
+	clientsMu      sync.RWMutex
 	recorder       *deliveryRecorder
 	ack            *DurationHistogram
 	attempted      atomic.Int64
@@ -132,27 +133,59 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo
 		return err
 	}
 	defer stopMetrics()
-	if err = run.connect(ctx, accounts, manifest.GroupTopic, apiKey, password); err != nil {
-		run.close()
-		return err
+	for {
+		err = run.connect(ctx, accounts, manifest.GroupTopic, apiKey, password)
+		if err == nil {
+			break
+		}
+		if ctx.Err() != nil {
+			run.close()
+			return ctx.Err()
+		}
+		log.Printf("agent=%s initial connection set incomplete; retrying: %v", machine, err)
+		if err = waitContext(ctx, 2*time.Second); err != nil {
+			run.close()
+			return err
+		}
 	}
 	defer run.close()
-	start, err := coord.ready(ctx, AgentReady{Machine: machine, ReadyConnections: len(run.clients)})
+	maintenanceCtx, stopMaintenance := context.WithCancel(ctx)
+	maintenanceDone := make(chan struct{})
+	go func() {
+		defer close(maintenanceDone)
+		run.maintainConnections(maintenanceCtx, accounts, manifest.GroupTopic, apiKey, password)
+	}()
+	start, err := coord.ready(ctx, func() AgentReady {
+		return AgentReady{
+			Machine: machine, ReadyConnections: run.aliveConnections(), ExpectedConnections: len(accounts),
+		}
+	})
 	if err != nil {
+		stopMaintenance()
+		<-maintenanceDone
 		return err
 	}
-	if alive := run.aliveConnections(); alive != len(run.clients) {
-		return fmt.Errorf("connections alive after coordinator wait=%d expected=%d", alive, len(run.clients))
-	}
 	if !runIDPattern.MatchString(start.RunID) || !validScenario(start.Scenario) {
+		stopMaintenance()
+		<-maintenanceDone
 		return fmt.Errorf("invalid coordinator start: %+v", start)
 	}
 	run.runID = start.RunID
 	run.scenario = start.Scenario
 	run.recorder.setRunID(start.RunID)
 	if err = waitUntil(ctx, start.StartAt); err != nil {
+		stopMaintenance()
+		<-maintenanceDone
 		return err
 	}
+	stopMaintenance()
+	<-maintenanceDone
+	if alive := run.aliveConnections(); alive != len(accounts) {
+		return fmt.Errorf("connections alive at scheduled start=%d expected=%d", alive, len(accounts))
+	}
+	// Connection churn while waiting for the other machine is outside the load
+	// measurement window. From this point onward every disconnect is a failure.
+	run.recorder.resetDisconnects()
 	startedAt := time.Now().UTC()
 	log.Printf("agent=%s scenario=%s started", machine, start.Scenario)
 	if err = run.execute(ctx, start.StartAt, manifest.GroupTopic); err != nil {
@@ -177,7 +210,7 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo
 	base := AgentSummary{
 		RunID: start.RunID, Scenario: start.Scenario, Machine: machine, StartedAt: startedAt,
 		LoadEndedAt: loadEndedAt, FinishedAt: time.Now().UTC(), ExpectedConnections: cfg.UsersPerMachine,
-		ReadyConnections: len(run.clients),
+		ReadyConnections: run.clientCount(),
 		PublishAttempted: run.attempted.Load(), PublishAccepted: int64(len(local)), PublishRejected: run.rejected.Load(),
 		PublishErrors: run.publishErrorCounts(),
 		AckLatency:    run.ack.Snapshot(), AcceptedLocal: local,
@@ -245,9 +278,9 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag
 	queries := make([]historyQuery, 0, len(byTopic))
 	for topic, messages := range byTopic {
 		sort.Slice(messages, func(i, j int) bool { return messages[i].Seq < messages[j].Seq })
-		client := r.clients[messages[0].PeerIndex]
+		client := r.client(messages[0].PeerIndex)
 		if messages[0].Kind == "group" {
-			client = r.clients[1]
+			client = r.client(1)
 		}
 		if client == nil {
 			r.addError(fmt.Sprintf("history topic=%s has no verifier client", topicHash(topic)))
@@ -293,9 +326,9 @@ func (r *agentRun) historyQueriesForMessages(messages []AcceptedMessage) []histo
 	queries := make([]historyQuery, 0, len(byTopic))
 	for topic, topicMessages := range byTopic {
 		sort.Slice(topicMessages, func(i, j int) bool { return topicMessages[i].Seq < topicMessages[j].Seq })
-		client := r.clients[topicMessages[0].PeerIndex]
+		client := r.client(topicMessages[0].PeerIndex)
 		if topicMessages[0].Kind == "group" {
-			client = r.clients[1]
+			client = r.client(1)
 		}
 		if client == nil {
 			continue
@@ -372,9 +405,14 @@ func (r *agentRun) connect(ctx context.Context, accounts []AccountRecord, groupT
 	interval := ramp / time.Duration(len(accounts))
 	var wg sync.WaitGroup
 	errCh := make(chan error, len(accounts))
-	var mu sync.Mutex
 	for _, account := range accounts {
 		account := account
+		if client := r.client(account.Index); client != nil {
+			if client.alive() {
+				continue
+			}
+			client.Close()
+		}
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
@@ -383,9 +421,7 @@ func (r *agentRun) connect(ctx context.Context, accounts []AccountRecord, groupT
 				errCh <- connectErr
 				return
 			}
-			mu.Lock()
-			r.clients[account.Index] = client
-			mu.Unlock()
+			r.setClient(account.Index, client)
 			r.ready.Add(1)
 		}()
 		if interval > 0 {
@@ -402,12 +438,55 @@ func (r *agentRun) connect(ctx context.Context, accounts []AccountRecord, groupT
 			failures = append(failures, connectErr.Error())
 		}
 	}
-	if len(r.clients) != len(accounts) {
-		return fmt.Errorf("ready connections=%d expected=%d failures=%v", len(r.clients), len(accounts), failures)
+	if r.aliveConnections() != len(accounts) {
+		return fmt.Errorf("ready connections=%d expected=%d failures=%v", r.aliveConnections(), len(accounts), failures)
 	}
 	return nil
 }

+func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRecord, groupTopic, apiKey, password string) {
+	byIndex := make(map[int]AccountRecord, len(accounts))
+	for _, account := range accounts {
+		byIndex[account.Index] = account
+	}
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		var wg sync.WaitGroup
+		for index, account := range byIndex {
+			current := r.client(index)
+			if current != nil && current.alive() {
+				continue
+			}
+			wg.Add(1)
+			go func(index int, account AccountRecord, previous *WSClient) {
+				defer wg.Done()
+				if previous != nil {
+					previous.Close()
+				}
+				client, err := r.connectAccountWithRetry(ctx, account, groupTopic, apiKey, password)
+				if err != nil {
+					if ctx.Err() == nil {
+						log.Printf("account=%s pre-start reconnect failed: %v", account.Username, err)
+					}
+					return
+				}
+				r.setClient(index, client)
+				log.Printf("account=%s pre-start connection restored", account.Username)
+			}(index, account, current)
+		}
+		wg.Wait()
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
 func (r *agentRun) connectAccountWithRetry(ctx context.Context, account AccountRecord, groupTopic, apiKey, password string) (*WSClient, error) {
 	var lastErr error
 	for attempt := 1; attempt <= 6; attempt++ {
@@ -467,7 +546,7 @@ func (r *agentRun) execute(ctx context.Context, start time.Time, groupTopic stri
 func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, messages int, interval time.Duration) error {
 	var wg sync.WaitGroup
 	for index := 1; index <= users; index++ {
-		client := r.clients[index]
+		client := r.client(index)
 		if client == nil {
 			return fmt.Errorf("missing client %d", index)
 		}
@@ -499,7 +578,7 @@ func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, mess
 }

 func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic string, first, last, rounds int) error {
-	if first < 1 || last > len(r.clients) || first > last {
+	if first < 1 || last > r.clientCount() || first > last {
 		return fmt.Errorf("invalid group publisher range %d-%d", first, last)
 	}
 	count := last - first + 1
@@ -510,7 +589,7 @@ func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic str
 	}
 	var wg sync.WaitGroup
 	for index := first; index <= last; index++ {
-		client := r.clients[index]
+		client := r.client(index)
 		position := index - first
 		wg.Add(1)
 		go func(index, position int, client *WSClient) {
@@ -591,7 +670,7 @@ func (r *agentRun) acceptedCopy() []AcceptedMessage {

 func (r *agentRun) aliveConnections() int {
 	alive := 0
-	for _, client := range r.clients {
+	for _, client := range r.clientsSnapshot() {
 		if client.alive() {
 			alive++
 		}
@@ -599,6 +678,34 @@ func (r *agentRun) aliveConnections() int {
 	return alive
 }

+func (r *agentRun) client(index int) *WSClient {
+	r.clientsMu.RLock()
+	defer r.clientsMu.RUnlock()
+	return r.clients[index]
+}
+
+func (r *agentRun) setClient(index int, client *WSClient) {
+	r.clientsMu.Lock()
+	r.clients[index] = client
+	r.clientsMu.Unlock()
+}
+
+func (r *agentRun) clientCount() int {
+	r.clientsMu.RLock()
+	defer r.clientsMu.RUnlock()
+	return len(r.clients)
+}
+
+func (r *agentRun) clientsSnapshot() []*WSClient {
+	r.clientsMu.RLock()
+	defer r.clientsMu.RUnlock()
+	clients := make([]*WSClient, 0, len(r.clients))
+	for _, client := range r.clients {
+		clients = append(clients, client)
+	}
+	return clients
+}
+
 func (r *agentRun) drainDuration() (time.Duration, error) {
 	if r.scenario == "p2p" {
 		return configDuration(r.cfg.DrainP2P)
@@ -607,7 +714,7 @@ func (r *agentRun) drainDuration() (time.Duration, error) {
 }

 func (r *agentRun) close() {
-	for _, client := range r.clients {
+	for _, client := range r.clientsSnapshot() {
 		client.Close()
 	}
 }
@@ -26,7 +26,7 @@ func startAgentMetrics(address string, run *agentRun) (func(), error) {
 		collectors.NewGoCollector(),
 		collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
 		prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "im_load_agent_ready_connections", Help: "Current ready WebSocket connections.", ConstLabels: labels}, func() float64 {
-			return float64(run.ready.Load())
+			return float64(run.aliveConnections())
 		}),
 		prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_attempted_total", Help: "Publish commands attempted.", ConstLabels: labels}, func() float64 {
 			return float64(run.attempted.Load())
@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+func TestMaintainConnectionsRestoresClosedClient(t *testing.T) {
+	var connections atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		connections.Add(1)
+		defer conn.Close()
+		for {
+			var packet map[string]json.RawMessage
+			if err = conn.ReadJSON(&packet); err != nil {
+				return
+			}
+			for _, raw := range packet {
+				var body struct {
+					ID string `json:"id"`
+				}
+				if json.Unmarshal(raw, &body) != nil || body.ID == "" {
+					continue
+				}
+				response := map[string]any{"ctrl": map[string]any{
+					"id": body.ID, "code": 200, "text": "ok",
+					"params": map[string]any{"user": "usrReconnectTest"},
+				}}
+				if err = conn.WriteJSON(response); err != nil {
+					return
+				}
+			}
+		}
+	}))
+	defer server.Close()
+
+	account := AccountRecord{
+		Machine: "a", Index: 1, Username: "lta0001", Token: "token",
+		P2PTopic: "p2pReconnectTest", Endpoint: "ws" + strings.TrimPrefix(server.URL, "http"),
+	}
+	cfg := &Config{Tenant: "allrich", CommandTimeout: "1s", WSPingInterval: "1h"}
+	run := &agentRun{
+		cfg: cfg, machine: "a", clients: make(map[int]*WSClient),
+		recorder: newDeliveryRecorder("pending", "a", 1), ack: NewDurationHistogram(),
+	}
+	initial, err := run.connectAccountWithRetry(context.Background(), account, "grpReconnectTest", "api-key", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	run.setClient(account.Index, initial)
+	initial.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	done := make(chan struct{})
+	go func() {
+		run.maintainConnections(ctx, []AccountRecord{account}, "grpReconnectTest", "api-key", "")
+		close(done)
+	}()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if replacement := run.client(account.Index); replacement != initial && replacement.alive() {
+			cancel()
+			<-done
+			run.close()
+			if got := connections.Load(); got < 2 {
+				t.Fatalf("websocket connections = %d, want at least 2", got)
+			}
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	<-done
+	run.close()
+	t.Fatal("closed client was not restored before deadline")
+}
@@ -16,22 +16,27 @@ import (
 )

 type StartInfo struct {
-	RunID    string    `json:"run_id"`
-	Scenario string    `json:"scenario"`
-	StartAt  time.Time `json:"start_at"`
+	RunID     string    `json:"run_id"`
+	Scenario  string    `json:"scenario"`
+	StartAt   time.Time `json:"start_at"`
+	Committed bool      `json:"committed"`
 }

 type AgentReady struct {
-	Machine          string `json:"machine"`
-	ReadyConnections int    `json:"ready_connections"`
+	Machine             string `json:"machine"`
+	ReadyConnections    int    `json:"ready_connections"`
+	ExpectedConnections int    `json:"expected_connections"`
+	Armed               bool   `json:"armed"`
 }

 type coordinatorState struct {
 	mu             sync.Mutex
 	start          StartInfo
 	startDelay     time.Duration
+	commitDelay    time.Duration
 	expectedAgents int
 	ready          map[string]AgentReady
+	armed          map[string]bool
 	ledgers        map[string][]AcceptedMessage
 	summaries      map[string]AgentSummary
 	complete       chan struct{}
@@ -49,9 +54,10 @@ func runCoordinator(ctx context.Context, listen, runID, scenario, output string,
 		expectedAgents = 2
 	}
 	state := &coordinatorState{
-		start: StartInfo{RunID: runID, Scenario: scenario}, startDelay: startDelay, expectedAgents: expectedAgents,
-		ready: make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage),
-		summaries: make(map[string]AgentSummary), complete: make(chan struct{}),
+		start: StartInfo{RunID: runID, Scenario: scenario}, startDelay: startDelay, commitDelay: 5 * time.Second,
+		expectedAgents: expectedAgents,
+		ready:          make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage),
+		armed: make(map[string]bool), summaries: make(map[string]AgentSummary), complete: make(chan struct{}),
 	}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/v1/ready", state.handleReady)
@@ -91,17 +97,48 @@ func (s *coordinatorState) handleReady(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	var ready AgentReady
-	if !decodeLimitedJSON(w, r, &ready) || (ready.Machine != "a" && ready.Machine != "b") || ready.ReadyConnections <= 0 {
+	if !decodeLimitedJSON(w, r, &ready) || (ready.Machine != "a" && ready.Machine != "b") ||
+		ready.ExpectedConnections <= 0 || ready.ReadyConnections < 0 || ready.ReadyConnections > ready.ExpectedConnections {
 		if ready.Machine != "a" && ready.Machine != "b" {
 			http.Error(w, "invalid machine", http.StatusBadRequest)
+		} else {
+			http.Error(w, "invalid connection counts", http.StatusBadRequest)
 		}
 		return
 	}
 	s.mu.Lock()
 	s.ready[ready.Machine] = ready
-	if len(s.ready) == s.expectedAgents && s.start.StartAt.IsZero() {
-		s.start.StartAt = time.Now().UTC().Add(s.startDelay)
-		log.Printf("all agents ready; start_at=%s", s.start.StartAt.Format(time.RFC3339Nano))
+	now := time.Now().UTC()
+	allStable := len(s.ready) == s.expectedAgents
+	for _, agent := range s.ready {
+		if agent.ReadyConnections != agent.ExpectedConnections {
+			allStable = false
+			break
+		}
+	}
+	if !allStable && (!s.start.Committed || now.Before(s.start.StartAt)) {
+		log.Printf("agent readiness dropped before start; resetting start barrier")
+		s.start.StartAt = time.Time{}
+		s.start.Committed = false
+		clear(s.armed)
+	}
+	if allStable && s.start.StartAt.IsZero() {
+		s.start.StartAt = now.Add(s.startDelay)
+		s.start.Committed = false
+		clear(s.armed)
+		log.Printf("all agents ready; validation_at=%s", s.start.StartAt.Format(time.RFC3339Nano))
+	}
+	if allStable && !s.start.Committed && ready.Armed && !now.Before(s.start.StartAt) {
+		s.armed[ready.Machine] = true
+	}
+	if allStable && !s.start.Committed && len(s.armed) == s.expectedAgents {
+		delay := s.commitDelay
+		if delay <= 0 {
+			delay = 5 * time.Second
+		}
+		s.start.StartAt = now.Add(delay)
+		s.start.Committed = true
+		log.Printf("all agents validated; committed start_at=%s", s.start.StartAt.Format(time.RFC3339Nano))
 	}
 	start := s.start
 	s.mu.Unlock()
@@ -185,7 +222,8 @@ func (s *coordinatorState) handleStatus(w http.ResponseWriter, _ *http.Request)
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	writeJSON(w, http.StatusOK, map[string]any{
-		"start": s.start, "ready": s.ready, "ledger_agents": len(s.ledgers), "summary_agents": len(s.summaries),
+		"start": s.start, "ready": s.ready, "armed": s.armed,
+		"ledger_agents": len(s.ledgers), "summary_agents": len(s.summaries),
 	})
 }

@@ -232,24 +270,37 @@ func newCoordinatorClient(base string) *coordinatorClient {
 	return &coordinatorClient{base: strings.TrimRight(base, "/"), client: &http.Client{Timeout: 35 * time.Second}}
 }

-func (c *coordinatorClient) ready(ctx context.Context, ready AgentReady) (StartInfo, error) {
+func (c *coordinatorClient) ready(ctx context.Context, current func() AgentReady) (StartInfo, error) {
 	var start StartInfo
-	if _, err := c.request(ctx, http.MethodPost, "/v1/ready", ready, &start); err != nil {
-		return start, err
-	}
-	for start.StartAt.IsZero() {
+	armed := false
+	for {
+		state := current()
+		state.Armed = armed
+		if _, err := c.request(ctx, http.MethodPost, "/v1/ready", state, &start); err != nil {
+			return start, err
+		}
+		if start.StartAt.IsZero() {
+			armed = false
+		} else {
+			remaining := time.Until(start.StartAt)
+			if start.Committed && remaining <= 0 {
+				return start, nil
+			}
+			if !start.Committed && remaining <= 0 {
+				armed = true
+				continue
+			}
+			if remaining < time.Second {
+				if err := waitContext(ctx, remaining); err != nil {
+					return start, err
+				}
+				continue
+			}
+		}
 		if err := waitContext(ctx, time.Second); err != nil {
 			return start, err
 		}
-		status, err := c.request(ctx, http.MethodGet, "/v1/start", nil, &start)
-		if err != nil {
-			return start, err
-		}
-		if status != http.StatusOK {
-			start = StartInfo{}
-		}
 	}
-	return start, nil
 }

 func (c *coordinatorClient) submitLedger(ctx context.Context, machine string, messages []AcceptedMessage) error {
@@ -1,18 +1,57 @@
 package main

 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 	"time"
 )

+func TestCoordinatorResetsBarrierWhenConnectionCountDrops(t *testing.T) {
+	state := &coordinatorState{
+		start: StartInfo{RunID: "run-reset", Scenario: "group"}, startDelay: time.Minute,
+		commitDelay: time.Second, expectedAgents: 2, ready: make(map[string]AgentReady),
+		armed: make(map[string]bool), ledgers: make(map[string][]AcceptedMessage),
+		summaries: make(map[string]AgentSummary), complete: make(chan struct{}),
+	}
+	postReady := func(ready AgentReady) StartInfo {
+		raw, err := json.Marshal(ready)
+		if err != nil {
+			t.Fatal(err)
+		}
+		req := httptest.NewRequest(http.MethodPost, "/v1/ready", bytes.NewReader(raw))
+		w := httptest.NewRecorder()
+		state.handleReady(w, req)
+		if w.Code != http.StatusOK {
+			t.Fatalf("ready status=%d body=%s", w.Code, w.Body.String())
+		}
+		var start StartInfo
+		if err = json.Unmarshal(w.Body.Bytes(), &start); err != nil {
+			t.Fatal(err)
+		}
+		return start
+	}
+	fullA := AgentReady{Machine: "a", ReadyConnections: 1000, ExpectedConnections: 1000}
+	fullB := AgentReady{Machine: "b", ReadyConnections: 1000, ExpectedConnections: 1000}
+	postReady(fullA)
+	if start := postReady(fullB); start.StartAt.IsZero() {
+		t.Fatal("full agents must create a validation barrier")
+	}
+	fullA.ReadyConnections = 999
+	if start := postReady(fullA); !start.StartAt.IsZero() || start.Committed {
+		t.Fatalf("connection drop must reset barrier: %+v", start)
+	}
+}
+
 func TestCoordinatorClientWaitsForBothAgentsAndMergesLedger(t *testing.T) {
 	state := &coordinatorState{
-		start: StartInfo{RunID: "run1", Scenario: "p2p"}, startDelay: time.Second, expectedAgents: 2,
+		start: StartInfo{RunID: "run1", Scenario: "p2p"}, startDelay: 50 * time.Millisecond,
+		commitDelay: 50 * time.Millisecond, expectedAgents: 2,
 		ready: make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage),
-		summaries: make(map[string]AgentSummary), complete: make(chan struct{}),
+		armed: make(map[string]bool), summaries: make(map[string]AgentSummary), complete: make(chan struct{}),
 	}
 	server := httptest.NewServer(testCoordinatorMux(state))
 	defer server.Close()
@@ -20,12 +59,16 @@ func TestCoordinatorClientWaitsForBothAgentsAndMergesLedger(t *testing.T) {

 	aDone := make(chan StartInfo, 1)
 	go func() {
-		start, _ := client.ready(context.Background(), AgentReady{Machine: "a", ReadyConnections: 1000})
+		start, _ := client.ready(context.Background(), func() AgentReady {
+			return AgentReady{Machine: "a", ReadyConnections: 1000, ExpectedConnections: 1000}
+		})
 		aDone <- start
 	}()
 	time.Sleep(50 * time.Millisecond)
-	startB, err := client.ready(context.Background(), AgentReady{Machine: "b", ReadyConnections: 1000})
-	if err != nil || startB.StartAt.IsZero() {
+	startB, err := client.ready(context.Background(), func() AgentReady {
+		return AgentReady{Machine: "b", ReadyConnections: 1000, ExpectedConnections: 1000}
+	})
+	if err != nil || startB.StartAt.IsZero() || !startB.Committed {
 		t.Fatalf("machine B ready: start=%+v err=%v", startB, err)
 	}
 	select {
@@ -54,6 +54,8 @@ func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorde

 func (r *deliveryRecorder) setRunID(runID string) { r.runID.Store(runID) }

+func (r *deliveryRecorder) resetDisconnects() { r.disconnects.Store(0) }
+
 func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) {
 	runID, _ := r.runID.Load().(string)
 	mid, _ := data.Head["x-im-client-mid"].(string)