test: support staged pressure connection counts

loadtest/im-pressure/agent.go

@@ -173,7 +173,8 @@ 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(), ReadyConnections: len(run.clients),
+		LoadEndedAt: loadEndedAt, FinishedAt: time.Now().UTC(), ExpectedConnections: cfg.UsersPerMachine,
+		ReadyConnections: len(run.clients),
 		PublishAttempted: run.attempted.Load(), PublishAccepted: int64(len(local)), PublishRejected: run.rejected.Load(),
 		PublishErrors: run.publishErrorCounts(),
 		AckLatency:    run.ack.Snapshot(), AcceptedLocal: local,

loadtest/im-pressure/merge.go

@@ -14,6 +14,7 @@ type CombinedReport struct {
 	Scenario            string            `json:"scenario"`
 	GeneratedAt         time.Time         `json:"generated_at"`
 	Agents              []AgentSummary    `json:"agents"`
+	ExpectedConnections int               `json:"expected_connections"`
 	ReadyConnections    int               `json:"ready_connections"`
 	PublishAttempted    int64             `json:"publish_attempted"`
 	PublishAccepted     int64             `json:"publish_accepted"`
@@ -54,6 +55,11 @@ func runMerge(aPath, bPath, output string) error {
 	report := CombinedReport{RunID: a.RunID, Scenario: a.Scenario, GeneratedAt: time.Now().UTC(), Agents: []AgentSummary{a, b}, Passed: true,
 		PublishErrors: make(map[string]int64)}
 	for _, summary := range report.Agents {
+		expectedConnections := summary.ExpectedConnections
+		if expectedConnections == 0 {
+			expectedConnections = 1000
+		}
+		report.ExpectedConnections += expectedConnections
 		report.ReadyConnections += summary.ReadyConnections
 		report.PublishAttempted += summary.PublishAttempted
 		report.PublishAccepted += summary.PublishAccepted
@@ -85,8 +91,9 @@ func runMerge(aPath, bPath, output string) error {
 		report.ArrivalRate = float64(report.ReceivedDeliveries) / float64(report.ExpectedDeliveries)
 		report.LossRate = float64(report.RealtimeLost) / float64(report.ExpectedDeliveries)
 	}
-	if report.ReadyConnections != 2000 {
-		report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ready connections=%d expected=2000", report.ReadyConnections))
+	if report.ReadyConnections != report.ExpectedConnections {
+		report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ready connections=%d expected=%d",
+			report.ReadyConnections, report.ExpectedConnections))
 	}
 	if report.PublishAttempted > 0 && report.PublishSuccessRate < 0.9999 {
 		report.FailureReasons = append(report.FailureReasons,

loadtest/im-pressure/merge_test.go

@@ -10,9 +10,11 @@ import (
 func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) {
 	dir := t.TempDir()
 	a := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "a", ReadyConnections: 1000,
+		ExpectedConnections: 1000,
 		PublishAttempted:    10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 9,
 		RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10}
 	b := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "b", ReadyConnections: 1000,
+		ExpectedConnections: 1000,
 		PublishAttempted:    10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10,
 		HistoryChecked: 10, HistoryFound: 9, PersistenceLost: 1}
 	aPath, bPath, reportPath := filepath.Join(dir, "a.json"), filepath.Join(dir, "b.json"), filepath.Join(dir, "report.json")

loadtest/im-pressure/model.go

@@ -349,6 +349,7 @@ type AgentSummary struct {
 	StartedAt           time.Time         `json:"started_at"`
 	LoadEndedAt         time.Time         `json:"load_ended_at"`
 	FinishedAt          time.Time         `json:"finished_at"`
+	ExpectedConnections int               `json:"expected_connections"`
 	ReadyConnections    int               `json:"ready_connections"`
 	PublishAttempted    int64             `json:"publish_attempted"`
 	PublishAccepted     int64             `json:"publish_accepted"`