From 622fc504180daefd8c6844c9854d0e4312e1d91a Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Mon, 28 Sep 2026 00:16:51 +0800 Subject: [PATCH] fix: keep load agents connected before start --- loadtest/im-pressure/MACHINE-B.md | 7 +- loadtest/im-pressure/README.md | 2 + loadtest/im-pressure/agent.go | 153 ++++++++++++++++--- loadtest/im-pressure/agent_metrics.go | 2 +- loadtest/im-pressure/agent_reconnect_test.go | 88 +++++++++++ loadtest/im-pressure/coordinator.go | 103 +++++++++---- loadtest/im-pressure/coordinator_test.go | 53 ++++++- loadtest/im-pressure/recorder.go | 2 + 8 files changed, 352 insertions(+), 58 deletions(-) create mode 100644 loadtest/im-pressure/agent_reconnect_test.go diff --git a/loadtest/im-pressure/MACHINE-B.md b/loadtest/im-pressure/MACHINE-B.md index 72f86aa..000b1ee 100644 --- a/loadtest/im-pressure/MACHINE-B.md +++ b/loadtest/im-pressure/MACHINE-B.md @@ -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 决定。 diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index 13bef67..d46450b 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -42,6 +42,8 @@ - `8090` 是无鉴权的压测协调端口，只允许压测内网访问，不要暴露到公网。 - 两台机器必须使用 NTP/chrony 同步时间。报告会分别输出同机时钟延迟和跨机器单向延迟；跨机器出现负延迟样本时，后者仅用于诊断，不参与门禁。 - agent 默认每 20 秒发送 WebSocket Ping，覆盖等待另一台机器、发压、drain 和历史核验阶段；`ws_ping_interval` 必须小于公网入口的空闲连接超时。 +- 正式开始前 agent 会持续检查实际存活连接；连接断开后自动重新登录并订阅 P2P、群 Topic。初次连接不齐或等待另一台机器期间掉线不会退出。 +- coordinator 使用两阶段屏障：两边实时连接数都达到配置值后进入验证倒计时；倒计时内任一方掉线就撤销屏障。两边再次确认满连接后才提交最终开始时间。 ## 统计口径 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index a58006c..1264e72 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -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() } } diff --git a/loadtest/im-pressure/agent_metrics.go b/loadtest/im-pressure/agent_metrics.go index 45cafa5..1aac369 100644 --- a/loadtest/im-pressure/agent_metrics.go +++ b/loadtest/im-pressure/agent_metrics.go @@ -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()) diff --git a/loadtest/im-pressure/agent_reconnect_test.go b/loadtest/im-pressure/agent_reconnect_test.go new file mode 100644 index 0000000..e6eafc4 --- /dev/null +++ b/loadtest/im-pressure/agent_reconnect_test.go @@ -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") +} diff --git a/loadtest/im-pressure/coordinator.go b/loadtest/im-pressure/coordinator.go index 4cb0c65..c9befda 100644 --- a/loadtest/im-pressure/coordinator.go +++ b/loadtest/im-pressure/coordinator.go @@ -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 { diff --git a/loadtest/im-pressure/coordinator_test.go b/loadtest/im-pressure/coordinator_test.go index 0db3bed..8caf0cc 100644 --- a/loadtest/im-pressure/coordinator_test.go +++ b/loadtest/im-pressure/coordinator_test.go @@ -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 { diff --git a/loadtest/im-pressure/recorder.go b/loadtest/im-pressure/recorder.go index 3c24140..88f2036 100644 --- a/loadtest/im-pressure/recorder.go +++ b/loadtest/im-pressure/recorder.go @@ -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)