From d501f23473cba663e9167fed029f092a638f44dd Mon Sep 17 00:00:00 2001 From: naiba Date: Mon, 20 Jul 2026 04:54:03 +0000 Subject: [PATCH] test(agentcompat): add exact stress scenario Co-authored-by: naiba/CloudCode --- .../internal/scenario/stress_artifact.go | 188 +++++++++++++++++ .../internal/scenario/stress_artifact_test.go | 96 +++++++++ .../internal/scenario/stress_evidence.go | 169 +++++++++++++++ .../internal/scenario/stress_evidence_test.go | 147 +++++++++++++ .../scenario/stress_evidence_validation.go | 154 ++++++++++++++ .../internal/scenario/stress_exact_once.go | 100 +++++++++ .../scenario/stress_exact_once_test.go | 43 ++++ .../internal/scenario/stress_identity.go | 185 ++++++++++++++++ .../internal/scenario/stress_path_lock.go | 73 +++++++ .../scenario/stress_path_lock_test.go | 35 ++++ .../internal/scenario/stress_plan.go | 120 +++++++++++ .../internal/scenario/stress_plan_test.go | 109 ++++++++++ .../scenario/stress_qa_artifact_test.go | 36 ++++ .../internal/scenario/stress_real_e2e_test.go | 197 ++++++++++++++++++ .../scenario/stress_real_operations.go | 152 ++++++++++++++ .../scenario/stress_real_operations_test.go | 78 +++++++ .../internal/scenario/stress_resource.go | 108 ++++++++++ .../internal/scenario/stress_resource_test.go | 178 ++++++++++++++++ .../internal/scenario/stress_round.go | 90 ++++++++ .../internal/scenario/stress_round_run.go | 91 ++++++++ .../internal/scenario/stress_soak.go | 88 ++++++++ .../internal/scenario/stress_soak_test.go | 86 ++++++++ .../internal/scenario/stress_sqlite_drain.go | 87 ++++++++ .../scenario/stress_sqlite_drain_test.go | 174 ++++++++++++++++ 24 files changed, 2784 insertions(+) create mode 100644 integration/agentcompat/internal/scenario/stress_artifact.go create mode 100644 integration/agentcompat/internal/scenario/stress_artifact_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_evidence.go create mode 100644 integration/agentcompat/internal/scenario/stress_evidence_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_evidence_validation.go create mode 100644 integration/agentcompat/internal/scenario/stress_exact_once.go create mode 100644 integration/agentcompat/internal/scenario/stress_exact_once_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_identity.go create mode 100644 integration/agentcompat/internal/scenario/stress_path_lock.go create mode 100644 integration/agentcompat/internal/scenario/stress_path_lock_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_plan.go create mode 100644 integration/agentcompat/internal/scenario/stress_plan_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_qa_artifact_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_real_e2e_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_real_operations.go create mode 100644 integration/agentcompat/internal/scenario/stress_real_operations_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_resource.go create mode 100644 integration/agentcompat/internal/scenario/stress_resource_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_round.go create mode 100644 integration/agentcompat/internal/scenario/stress_round_run.go create mode 100644 integration/agentcompat/internal/scenario/stress_soak.go create mode 100644 integration/agentcompat/internal/scenario/stress_soak_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_sqlite_drain.go create mode 100644 integration/agentcompat/internal/scenario/stress_sqlite_drain_test.go diff --git a/integration/agentcompat/internal/scenario/stress_artifact.go b/integration/agentcompat/internal/scenario/stress_artifact.go new file mode 100644 index 00000000..cc8489e4 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_artifact.go @@ -0,0 +1,188 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" + "github.com/nezhahq/nezha/integration/agentcompat/internal/evidence" +) + +var ErrStressArtifactInvalid = errors.New("stress artifact is invalid") + +type stressArtifact struct { + Version int `json:"version"` + Profile string `json:"profile"` + Seed string `json:"seed"` + RoundCount int `json:"round_count"` + OperationCount int `json:"operation_count"` + SessionCount int `json:"session_count"` + WarmupCount int `json:"warmup_count"` + ResourceSummaryCount int `json:"resource_summary_count"` + ResourceSampleCount int `json:"resource_sample_count"` + ResourceIntervalMilliseconds int64 `json:"resource_interval_milliseconds"` + Quotas StressQuotaEvidence `json:"quotas"` + PathLockStripes int `json:"path_lock_stripes"` + DuplicateOperations int `json:"duplicate_operations"` + ResourceDrift int `json:"resource_drift"` + RSSBounded bool `json:"rss_bounded"` + Cleanup StressCleanupSummary `json:"cleanup"` +} + +func stressPRFullProfile() (contract.Profile, error) { + return contract.ProfileByName(string(contract.ProfilePRFull)) +} + +func publishStressEvidence(root string, value StressEvidence) error { + profile, err := stressPRFullProfile() + if err != nil { + return err + } + if err := value.ValidateSuccess(profile); err != nil { + return err + } + artifact, err := newStressArtifact(value) + if err != nil { + return err + } + data, err := json.Marshal(artifact) + if err != nil { + return err + } + if evidence.Redact(string(data)) != string(data) { + return ErrStressArtifactInvalid + } + info, err := os.Lstat(root) + if errors.Is(err, os.ErrNotExist) { + if err := os.Mkdir(root, 0o700); err != nil { + return err + } + info, err = os.Lstat(root) + } + if err != nil || info.Mode()&os.ModeSymlink != 0 || !info.IsDir() || info.Mode().Perm() != 0o700 { + return ErrStressArtifactInvalid + } + path := filepath.Join(root, "stress.json") + if stale, statErr := os.Lstat(path); statErr == nil { + if stale.Mode()&os.ModeSymlink != 0 || !stale.Mode().IsRegular() { + return ErrStressArtifactInvalid + } + } else if !errors.Is(statErr, os.ErrNotExist) { + return statErr + } + temporary, err := os.CreateTemp(root, ".stress-*") + if err != nil { + return err + } + temporaryName := temporary.Name() + defer os.Remove(temporaryName) + if err := temporary.Chmod(0o600); err != nil { + _ = temporary.Close() + return err + } + if _, err := temporary.Write(data); err != nil { + _ = temporary.Close() + return err + } + if err := temporary.Close(); err != nil { + return err + } + return os.Rename(temporaryName, path) +} + +func readStressEvidence(root string) (stressArtifact, error) { + var value stressArtifact + path := filepath.Join(root, "stress.json") + info, err := os.Lstat(path) + if err != nil { + return value, err + } + if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() || info.Mode().Perm() != 0o600 { + return value, ErrStressArtifactInvalid + } + data, err := os.ReadFile(path) + if err != nil { + return value, err + } + if evidence.Redact(string(data)) != string(data) { + return value, ErrStressArtifactInvalid + } + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&value); err != nil { + return value, fmt.Errorf("decode stress artifact: %w", ErrStressArtifactInvalid) + } + var trailing struct{} + if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) { + return value, ErrStressArtifactInvalid + } + if err := value.validate(); err != nil { + return value, err + } + return value, nil +} + +func newStressArtifact(value StressEvidence) (stressArtifact, error) { + profile, err := contract.ProfileByName(string(value.Profile)) + if err != nil { + return stressArtifact{}, err + } + artifact := stressArtifact{Version: 1, Profile: string(value.Profile), Seed: fmt.Sprintf("%08x", uint64(value.Seed)), SessionCount: len(value.Sessions), WarmupCount: len(value.Warmups), ResourceSummaryCount: 1 + profile.AgentCount(), ResourceSampleCount: 5, ResourceIntervalMilliseconds: 250, Quotas: value.Quotas, PathLockStripes: value.Quotas.PathLockStripes, Cleanup: value.Cleanup} + operationIDs := make(map[StressOperationID]struct{}) + resourceEvaluations := make([]StressResourceEvaluation, 0, artifact.ResourceSummaryCount) + for _, iteration := range value.Iterations { + artifact.RoundCount += len(iteration.Rounds) + for _, round := range iteration.Rounds { + artifact.OperationCount += len(round.Operations) + for _, operation := range round.Operations { + if _, duplicate := operationIDs[operation.ID]; duplicate { + artifact.DuplicateOperations++ + } + operationIDs[operation.ID] = struct{}{} + } + } + for _, resource := range iteration.Resources { + evaluation, err := EvaluateStressResource(resource) + if err != nil { + return stressArtifact{}, err + } + resourceEvaluations = append(resourceEvaluations, evaluation) + } + } + resourceDrift, rssBounded, err := aggregateStressResourceEvaluations(resourceEvaluations, artifact.ResourceSummaryCount) + if err != nil { + return stressArtifact{}, err + } + artifact.ResourceDrift = resourceDrift + artifact.RSSBounded = rssBounded + return artifact, artifact.validate() +} + +func aggregateStressResourceEvaluations(evaluations []StressResourceEvaluation, expectedCount int) (int, bool, error) { + if len(evaluations) != expectedCount { + return 0, false, fmt.Errorf("resource evaluations=%d want=%d: %w", len(evaluations), expectedCount, ErrStressArtifactInvalid) + } + drift := 0 + rssBounded := true + for _, evaluation := range evaluations { + if evaluation.Baseline.Descendants != evaluation.End.Descendants || evaluation.Baseline.NonStdioFDs != evaluation.End.NonStdioFDs || evaluation.Baseline.TCPListeners != evaluation.End.TCPListeners || evaluation.Baseline.TCP6Listeners != evaluation.End.TCP6Listeners { + drift++ + } + rssBounded = rssBounded && evaluation.RSSDeltaBytes <= evaluation.RSSLimitBytes + } + return drift, rssBounded, nil +} + +func (artifact stressArtifact) validate() error { + if artifact.Version != 1 || artifact.Profile != string(contract.ProfilePRFull) || artifact.Seed != "4e5a4841" || artifact.RoundCount != 4 || artifact.OperationCount != 64 || artifact.SessionCount != 12 || artifact.WarmupCount != 8 || artifact.ResourceSummaryCount != 9 || artifact.ResourceSampleCount != 5 || artifact.ResourceIntervalMilliseconds != 250 || artifact.PathLockStripes != 1024 || artifact.DuplicateOperations != 0 || artifact.ResourceDrift != 0 || !artifact.RSSBounded || !artifact.Cleanup.Passed || artifact.Cleanup.ReceiptCount != 9 || artifact.Cleanup.FailedReceiptCount != 0 || artifact.Cleanup.ForcedCleanupCount != 0 || artifact.Cleanup.ProcessResidue != 0 || artifact.Cleanup.ProcessGroupResidue != 0 || artifact.Cleanup.WorkspaceResidue != 0 { + return ErrStressArtifactInvalid + } + return artifact.Quotas.Validate() +} diff --git a/integration/agentcompat/internal/scenario/stress_artifact_test.go b/integration/agentcompat/internal/scenario/stress_artifact_test.go new file mode 100644 index 00000000..1da80cdc --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_artifact_test.go @@ -0,0 +1,96 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +func TestStressArtifactRejectsUnknownAndRuntimeFields(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + root := t.TempDir() + valid := validStressArtifact(t, profile) + data, err := json.Marshal(valid) + require.NoError(t, err) + for _, mutation := range []string{ + `{"unknown":true}`, + `{"pid":123}`, + `{"operation_id":"op"}`, + } { + path := filepath.Join(root, "stress.json") + require.NoError(t, os.WriteFile(path, append(data[:len(data)-1], []byte(","+mutation[1:])...), 0o600)) + _, err = readStressEvidence(root) + require.ErrorIs(t, err, ErrStressArtifactInvalid) + } +} + +func TestStressArtifactRejectsTrailingDataAndWrongSeed(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + root := t.TempDir() + valid := validStressArtifact(t, profile) + data, err := json.Marshal(valid) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(root, "stress.json"), append(data, []byte("\n{}")...), 0o600)) + _, err = readStressEvidence(root) + require.ErrorIs(t, err, ErrStressArtifactInvalid) + + valid.Seed = "4e5a4842" + data, err = json.Marshal(valid) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(root, "stress.json"), data, 0o600)) + _, err = readStressEvidence(root) + require.ErrorIs(t, err, ErrStressArtifactInvalid) +} + +func TestStressCleanupMutationsCannotPublishSuccess(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + value := validStressEvidence(t, profile) + for _, mutate := range []func(*StressEvidence){ + func(evidence *StressEvidence) { evidence.Cleanup.FailedReceiptCount = 1 }, + func(evidence *StressEvidence) { evidence.Cleanup.ForcedCleanupCount = 1 }, + func(evidence *StressEvidence) { evidence.Cleanup.ProcessResidue = 1 }, + func(evidence *StressEvidence) { evidence.Cleanup.ProcessGroupResidue = 1 }, + func(evidence *StressEvidence) { evidence.Cleanup.WorkspaceResidue = 1 }, + } { + candidate := value + mutate(&candidate) + _, err := newStressArtifact(candidate) + require.Error(t, err) + } +} + +func TestStressArtifactAggregatesExactlyNineResourceEvaluations(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + value := validStressEvidence(t, profile) + artifact, err := newStressArtifact(value) + require.NoError(t, err) + require.Equal(t, 0, artifact.ResourceDrift) + require.True(t, artifact.RSSBounded) + + value.Iterations[0].Resources = value.Iterations[0].Resources[:8] + _, err = newStressArtifact(value) + require.ErrorIs(t, err, ErrStressArtifactInvalid) +} + +func validStressArtifact(t *testing.T, profile contract.Profile) stressArtifact { + t.Helper() + return stressArtifact{ + Version: 1, Profile: string(profile.Name()), Seed: "4e5a4841", + RoundCount: 4, OperationCount: 64, SessionCount: 12, WarmupCount: 8, + ResourceSummaryCount: 9, ResourceSampleCount: 5, ResourceIntervalMilliseconds: 250, + Quotas: StressQuotaEvidence{PATSecond: quotaBoundary(10, 11), PATMinute: quotaBoundary(120, 121), UserStreams: quotaBoundary(20, 21), ServerStreams: quotaBoundary(40, 41)}, + PathLockStripes: 1024, DuplicateOperations: 0, ResourceDrift: 0, RSSBounded: true, + Cleanup: StressCleanupSummary{Passed: true, ReceiptCount: 9}, + } +} diff --git a/integration/agentcompat/internal/scenario/stress_evidence.go b/integration/agentcompat/internal/scenario/stress_evidence.go new file mode 100644 index 00000000..8e7d9525 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_evidence.go @@ -0,0 +1,169 @@ +//go:build linux + +package scenario + +import ( + "errors" + "fmt" + "reflect" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +var ( + ErrStressEvidence = errors.New("stress evidence is invalid") + ErrStressFault = errors.New("stress worker fault evidence is invalid") +) + +type StressPreparedBinaries struct { + DashboardBuildCount int `json:"dashboard_build_count"` + DashboardPathReused bool `json:"dashboard_path_reused"` + AgentBuildCount int `json:"agent_build_count"` + AgentPathReused bool `json:"agent_path_reused"` +} + +type StressQuotaBoundary struct { + Allowed int `json:"allowed"` + Rejected int `json:"rejected"` + AllowedAccepted bool `json:"allowed_accepted"` + RejectedDenied bool `json:"rejected_denied"` +} + +type StressQuotaEvidence struct { + PATSecond StressQuotaBoundary `json:"pat_second"` + PATMinute StressQuotaBoundary `json:"pat_minute"` + UserStreams StressQuotaBoundary `json:"user_streams"` + ServerStreams StressQuotaBoundary `json:"server_streams"` + PathLockStripes int `json:"path_lock_stripes"` +} + +type StressWarmupEvidence struct { + Agent StressAgentOrdinal `json:"agent"` + Exec bool `json:"exec"` + Filesystem bool `json:"filesystem"` + Terminal bool `json:"terminal"` + NAT bool `json:"nat"` + FM bool `json:"file_manager"` +} + +type StressSessionEvidence struct { + ID StressSessionID `json:"id"` + Kind StressSessionKind `json:"kind"` + Succeeded bool `json:"succeeded"` +} + +type StressFaultTarget struct { + Iteration int `json:"iteration"` + Round int `json:"round"` + Agent StressAgentOrdinal `json:"agent"` + Kind StressOperationKind `json:"kind"` +} + +type StressCleanupSummary struct { + Passed bool `json:"passed"` + ReceiptCount int `json:"receipt_count"` + FailedReceiptCount int `json:"failed_receipt_count"` + ForcedCleanupCount int `json:"forced_cleanup_count"` + ProcessResidue int `json:"process_residue"` + ProcessGroupResidue int `json:"process_group_residue"` + WorkspaceResidue int `json:"workspace_residue"` +} + +type StressIterationEvidence struct { + Iteration int `json:"iteration"` + Rounds []StressRoundEvidence `json:"rounds"` + Resources []StressProcessWindows `json:"resources"` +} + +type StressEvidence struct { + Version int `json:"version"` + Profile contract.ProfileName `json:"profile"` + Seed contract.Seed `json:"seed"` + PreparedBinaries StressPreparedBinaries `json:"prepared_binaries"` + Quotas StressQuotaEvidence `json:"quotas"` + Warmups []StressWarmupEvidence `json:"warmups"` + Sessions []StressSessionEvidence `json:"sessions"` + Plan StressPlan `json:"plan"` + Iterations []StressIterationEvidence `json:"iterations"` + FaultTarget *StressFaultTarget `json:"fault_target,omitempty"` + SoakTrend StressSoakTrendEvidence `json:"soak_trend,omitempty"` + Cleanup StressCleanupSummary `json:"cleanup"` +} + +func StressWorkerFaultTarget() StressFaultTarget { + agent, err := NewStressAgentOrdinal(4) + if err != nil { + panic(err) + } + return StressFaultTarget{Iteration: 1, Round: 2, Agent: agent, Kind: StressOperationExec} +} + +func (e StressEvidence) ValidateSuccess(profile contract.Profile) error { + return e.validate(profile, false) +} + +func (e StressEvidence) ValidateStressWorker(profile contract.Profile) error { + return e.validate(profile, true) +} + +func (e StressEvidence) validate(profile contract.Profile, faultAware bool) error { + if e.Version != 1 || e.Profile != profile.Name() || e.Seed != contract.DefaultSeed { + return ErrStressEvidence + } + canonical, err := GenerateStressPlan(profile, e.Seed) + if err != nil || e.Plan.Profile != e.Profile || e.Plan.Seed != e.Seed || !reflect.DeepEqual(e.Plan, canonical) { + return fmt.Errorf("plan does not match canonical plan: %w", ErrStressEvidence) + } + if err := validateStressPreparedBinaries(e.PreparedBinaries); err != nil { + return err + } + if err := e.Quotas.Validate(); err != nil { + return err + } + if err := validateStressWarmups(e.Warmups, profile.AgentCount()); err != nil { + return err + } + if err := validateStressSessions(e.Sessions, canonical.Sessions, profile.ConcurrentSessions()); err != nil { + return err + } + if len(e.Iterations) != profile.Iterations() { + return fmt.Errorf("iterations=%d want=%d: %w", len(e.Iterations), profile.Iterations(), ErrStressEvidence) + } + failed := 0 + for index, iteration := range e.Iterations { + count, err := validateStressIteration(canonical, iteration, index+1, faultAware) + if err != nil { + return err + } + failed += count + } + if err := validateStressFault(e.FaultTarget, failed, faultAware); err != nil { + return err + } + if profile.Iterations() == 3 { + if err := ValidateStressSoakTrendForProfile(profile, e.SoakTrend); err != nil { + return err + } + } + if !e.Cleanup.Passed || e.Cleanup.ReceiptCount != 9 || e.Cleanup.FailedReceiptCount != 0 || e.Cleanup.ForcedCleanupCount != 0 || e.Cleanup.ProcessResidue != 0 || e.Cleanup.ProcessGroupResidue != 0 || e.Cleanup.WorkspaceResidue != 0 { + return fmt.Errorf("cleanup=%+v: %w", e.Cleanup, ErrStressEvidence) + } + return nil +} + +func (q StressQuotaEvidence) Validate() error { + wants := []struct { + got StressQuotaBoundary + allow int + reject int + }{{q.PATSecond, 10, 11}, {q.PATMinute, 120, 121}, {q.UserStreams, 20, 21}, {q.ServerStreams, 40, 41}} + for _, want := range wants { + if want.got.Allowed != want.allow || want.got.Rejected != want.reject || !want.got.AllowedAccepted || !want.got.RejectedDenied { + return fmt.Errorf("quota=%+v want=%d/%d: %w", want.got, want.allow, want.reject, ErrStressEvidence) + } + } + if q.PathLockStripes != 1024 { + return fmt.Errorf("path lock stripes=%d: %w", q.PathLockStripes, ErrStressEvidence) + } + return nil +} diff --git a/integration/agentcompat/internal/scenario/stress_evidence_test.go b/integration/agentcompat/internal/scenario/stress_evidence_test.go new file mode 100644 index 00000000..b4703b7a --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_evidence_test.go @@ -0,0 +1,147 @@ +//go:build linux + +package scenario + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +func TestStressEvidence_AcceptsCompleteSuccessContract(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + evidence := validStressEvidence(t, profile) + + err = evidence.ValidateSuccess(profile) + + require.NoError(t, err) +} + +func TestStressEvidence_AcceptsOnlyExactStressWorkerFault(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + evidence := validStressEvidence(t, profile) + target := StressWorkerFaultTarget() + evidence.FaultTarget = &target + faultOperation := findStressFaultOperation(evidence.Plan) + for roundIndex := range evidence.Iterations[0].Rounds { + for operationIndex := range evidence.Iterations[0].Rounds[roundIndex].Operations { + operation := &evidence.Iterations[0].Rounds[roundIndex].Operations[operationIndex] + if operation.ID == faultOperation.ID { + operation.Succeeded = false + operation.Error = "injected stress worker fault" + } + } + } + + err = evidence.ValidateStressWorker(profile) + + require.NoError(t, err) +} + +func TestStressEvidence_RejectsSecondFailedOperationForStressWorker(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + evidence := validStressEvidence(t, profile) + target := StressWorkerFaultTarget() + evidence.FaultTarget = &target + evidence.Iterations[0].Rounds[1].Operations[0].Succeeded = false + evidence.Iterations[0].Rounds[1].Operations[0].Error = "injected stress worker fault" + evidence.Iterations[0].Rounds[1].Operations[1].Succeeded = false + evidence.Iterations[0].Rounds[1].Operations[1].Error = "unexpected failure" + + err = evidence.ValidateStressWorker(profile) + + require.ErrorIs(t, err, ErrStressFault) +} + +func TestStressEvidence_RejectsSelfConsistentTruncatedPlan(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + evidence := validStressEvidence(t, profile) + evidence.Plan.Rounds = evidence.Plan.Rounds[:1] + evidence.Iterations[0].Rounds = evidence.Iterations[0].Rounds[:1] + + err = evidence.ValidateSuccess(profile) + + require.ErrorIs(t, err, ErrStressEvidence) +} + +func TestStressEvidence_RejectsDuplicateDashboardResources(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + evidence := validStressEvidence(t, profile) + evidence.Iterations[0].Resources[1] = evidence.Iterations[0].Resources[0] + + err = evidence.ValidateSuccess(profile) + + require.ErrorIs(t, err, ErrStressEvidence) +} + +func validStressEvidence(t *testing.T, profile contract.Profile) StressEvidence { + t.Helper() + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + warmups := make([]StressWarmupEvidence, profile.AgentCount()) + for index := range warmups { + agent, agentErr := NewStressAgentOrdinal(index + 1) + require.NoError(t, agentErr) + warmups[index] = StressWarmupEvidence{Agent: agent, Exec: true, Filesystem: true, Terminal: true, NAT: true, FM: true} + } + sessions := make([]StressSessionEvidence, len(plan.Sessions)) + for index, session := range plan.Sessions { + sessions[index] = StressSessionEvidence{ID: session.ID, Kind: session.Kind, Succeeded: true} + } + rounds := make([]StressRoundEvidence, len(plan.Rounds)) + started := time.Unix(100, 0) + for roundIndex, round := range plan.Rounds { + operations := make([]StressOperationEvidence, len(round.Operations)) + for operationIndex, operation := range round.Operations { + operations[operationIndex] = StressOperationEvidence{ID: operation.ID, Round: operation.Round, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind, LaunchedAt: started, CompletedAt: started.Add(time.Millisecond), Succeeded: true, SuccessProof: "ok"} + } + rounds[roundIndex] = StressRoundEvidence{Round: round.Round, Operations: operations} + } + resources := make([]StressProcessWindows, 0, profile.AgentCount()+1) + resources = append(resources, stressDashboardResourceFixture(100)) + for index := 1; index <= profile.AgentCount(); index++ { + agent, agentErr := NewStressAgentOrdinal(index) + require.NoError(t, agentErr) + process, processErr := NewStressAgentProcess(agent, 200+index) + require.NoError(t, processErr) + resources = append(resources, StressProcessWindows{Process: process, Baseline: stressWindow(200+index, 100), End: stressWindow(200+index, 100)}) + } + iterations := make([]StressIterationEvidence, profile.Iterations()) + for index := range iterations { + iterations[index] = StressIterationEvidence{Iteration: index + 1, Rounds: rounds, Resources: resources} + } + return StressEvidence{ + Version: 1, Profile: profile.Name(), Seed: contract.DefaultSeed, Plan: plan, + PreparedBinaries: StressPreparedBinaries{DashboardBuildCount: 1, DashboardPathReused: true, AgentBuildCount: 1, AgentPathReused: true}, + Quotas: StressQuotaEvidence{ + PATSecond: quotaBoundary(10, 11), PATMinute: quotaBoundary(120, 121), + UserStreams: quotaBoundary(20, 21), ServerStreams: quotaBoundary(40, 41), PathLockStripes: 1024, + }, + Warmups: warmups, Sessions: sessions, Iterations: iterations, + Cleanup: StressCleanupSummary{Passed: true, ReceiptCount: 9}, + } +} + +func quotaBoundary(allowed, rejected int) StressQuotaBoundary { + return StressQuotaBoundary{Allowed: allowed, Rejected: rejected, AllowedAccepted: true, RejectedDenied: true} +} + +func findStressFaultOperation(plan StressPlan) StressOperationPlan { + target := StressWorkerFaultTarget() + for _, round := range plan.Rounds { + for _, operation := range round.Operations { + if round.Round == target.Round && operation.Agent == target.Agent && operation.Kind == target.Kind { + return operation + } + } + } + panic("stress fault operation missing") +} diff --git a/integration/agentcompat/internal/scenario/stress_evidence_validation.go b/integration/agentcompat/internal/scenario/stress_evidence_validation.go new file mode 100644 index 00000000..afce3d54 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_evidence_validation.go @@ -0,0 +1,154 @@ +//go:build linux + +package scenario + +import "fmt" + +func validateStressPreparedBinaries(evidence StressPreparedBinaries) error { + if evidence.DashboardBuildCount != 1 || !evidence.DashboardPathReused || evidence.AgentBuildCount != 1 || !evidence.AgentPathReused { + return fmt.Errorf("prepared binaries=%+v: %w", evidence, ErrStressEvidence) + } + return nil +} + +func validateStressWarmups(warmups []StressWarmupEvidence, agentCount int) error { + if len(warmups) != agentCount { + return fmt.Errorf("warmups=%d want=%d: %w", len(warmups), agentCount, ErrStressEvidence) + } + seen := make(map[int]struct{}, len(warmups)) + for _, warmup := range warmups { + if _, duplicate := seen[warmup.Agent.Int()]; duplicate || !warmup.Exec || !warmup.Filesystem || !warmup.Terminal || !warmup.NAT || !warmup.FM { + return fmt.Errorf("warmup=%+v: %w", warmup, ErrStressEvidence) + } + seen[warmup.Agent.Int()] = struct{}{} + } + return nil +} + +func validateStressSessions(sessions []StressSessionEvidence, plan []StressSessionPlan, countPerKind int) error { + if len(sessions) != countPerKind*3 { + return fmt.Errorf("sessions=%d want=%d: %w", len(sessions), countPerKind*3, ErrStressEvidence) + } + if len(plan) != len(sessions) { + return fmt.Errorf("session plan=%d evidence=%d: %w", len(plan), len(sessions), ErrStressEvidence) + } + expected := make(map[StressSessionID]StressSessionPlan, len(plan)) + for _, session := range plan { + expected[session.ID] = session + } + counts := make(map[StressSessionKind]int, 3) + seen := make(map[StressSessionID]struct{}, len(sessions)) + for _, session := range sessions { + planned, exists := expected[session.ID] + _, duplicate := seen[session.ID] + if !exists || planned.Kind != session.Kind || duplicate || !session.Succeeded { + return fmt.Errorf("session=%+v: %w", session, ErrStressEvidence) + } + seen[session.ID] = struct{}{} + counts[session.Kind]++ + } + for _, kind := range []StressSessionKind{StressSessionTerminal, StressSessionNAT, StressSessionFM} { + if counts[kind] != countPerKind { + return fmt.Errorf("session kind=%s count=%d want=%d: %w", kind, counts[kind], countPerKind, ErrStressEvidence) + } + } + return nil +} + +func validateStressIteration(plan StressPlan, evidence StressIterationEvidence, iteration int, faultAware bool) (int, error) { + if evidence.Iteration != iteration || len(evidence.Rounds) != len(plan.Rounds) || len(evidence.Resources) != 1+canonicalAgentCount(plan) { + return 0, fmt.Errorf("iteration=%+v: %w", evidence, ErrStressEvidence) + } + failed := 0 + for index, round := range evidence.Rounds { + matched, err := matchStressRoundEvidence(plan.Rounds[index], round) + if err != nil { + return 0, err + } + for _, operation := range matched { + if !operation.Succeeded || operation.Error != "" { + failed++ + if !faultAware || !isStressFaultOperation(plan.Rounds[index], operation, iteration) { + return 0, fmt.Errorf("unexpected failed operation=%s: %w", operation.ID.String(), ErrStressFault) + } + } + } + } + if err := validateStressResourceIdentities(plan, evidence.Resources); err != nil { + return 0, err + } + for _, windows := range evidence.Resources { + if _, err := EvaluateStressResource(windows); err != nil { + return 0, err + } + } + return failed, nil +} + +func validateStressFault(target *StressFaultTarget, failed int, faultAware bool) error { + if !faultAware { + if target != nil || failed != 0 { + return ErrStressFault + } + return nil + } + want := StressWorkerFaultTarget() + if target == nil || *target != want || failed != 1 { + return fmt.Errorf("target=%+v failed=%d want=%+v/1: %w", target, failed, want, ErrStressFault) + } + return nil +} + +func isStressFaultOperation(plan StressRoundPlan, evidence StressOperationEvidence, iteration int) bool { + target := StressWorkerFaultTarget() + if iteration != target.Iteration || plan.Round != target.Round { + return false + } + for _, operation := range plan.Operations { + if operation.ID == evidence.ID { + return operation.Agent == target.Agent && operation.Kind == target.Kind + } + } + return false +} + +func planAgentCount(plan StressPlan) int { + return canonicalAgentCount(plan) +} + +func canonicalAgentCount(plan StressPlan) int { + if len(plan.Rounds) == 0 { + return 0 + } + return len(plan.Rounds[0].Operations) / 2 +} + +func validateStressResourceIdentities(plan StressPlan, resources []StressProcessWindows) error { + wantAgents := planAgentCount(plan) + if len(resources) != wantAgents+1 { + return fmt.Errorf("resources=%d want=%d: %w", len(resources), wantAgents+1, ErrStressEvidence) + } + dashboardCount := 0 + seenAgents := make(map[int]struct{}, wantAgents) + for _, resource := range resources { + switch resource.Process.Kind { + case StressProcessDashboard: + dashboardCount++ + case StressProcessAgent: + ordinal := resource.Process.Agent.Int() + if ordinal < 1 || ordinal > wantAgents { + return fmt.Errorf("unknown agent ordinal=%d: %w", ordinal, ErrStressEvidence) + } + if _, duplicate := seenAgents[ordinal]; duplicate { + return fmt.Errorf("duplicate agent ordinal=%d: %w", ordinal, ErrStressEvidence) + } + seenAgents[ordinal] = struct{}{} + default: + return fmt.Errorf("unknown process kind=%q: %w", resource.Process.Kind, ErrStressEvidence) + } + } + if dashboardCount != 1 || len(seenAgents) != wantAgents { + return fmt.Errorf("dashboard=%d agents=%d want=1/%d: %w", dashboardCount, len(seenAgents), wantAgents, ErrStressEvidence) + } + return nil +} diff --git a/integration/agentcompat/internal/scenario/stress_exact_once.go b/integration/agentcompat/internal/scenario/stress_exact_once.go new file mode 100644 index 00000000..b22ecdc7 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_exact_once.go @@ -0,0 +1,100 @@ +//go:build linux + +package scenario + +import ( + "errors" + "fmt" + "sync" +) + +var ( + ErrStressOperationUnknown = errors.New("stress operation is unknown") + ErrStressOperationDuplicateStart = errors.New("stress operation started more than once") + ErrStressOperationDuplicateCompletion = errors.New("stress operation completed more than once") + ErrStressOperationOwnerMismatch = errors.New("stress operation owner does not match plan") + ErrStressOperationMissingCompletion = errors.New("stress operation completion is missing") +) + +type StressOperationReceipt struct { + Operation StressOperationPlan `json:"operation"` + SuccessProof string `json:"success_proof"` +} + +type stressOperationState struct { + plan StressOperationPlan + started bool + completed bool +} + +type stressExactOnceRegistry struct { + mu sync.Mutex + operations map[StressOperationID]*stressOperationState +} + +func newStressExactOnceRegistry(plan StressPlan) (*stressExactOnceRegistry, error) { + operations := make(map[StressOperationID]*stressOperationState) + for _, round := range plan.Rounds { + for _, operation := range round.Operations { + if operation.ID.String() == "" { + return nil, fmt.Errorf("empty operation ID: %w", ErrStressOperationUnknown) + } + if _, exists := operations[operation.ID]; exists { + return nil, fmt.Errorf("duplicate operation ID %s: %w", operation.ID.String(), ErrStressOperationUnknown) + } + operations[operation.ID] = &stressOperationState{plan: operation} + } + } + return &stressExactOnceRegistry{operations: operations}, nil +} + +func (registry *stressExactOnceRegistry) Start(operation StressOperationPlan) error { + registry.mu.Lock() + defer registry.mu.Unlock() + state, exists := registry.operations[operation.ID] + if !exists { + return ErrStressOperationUnknown + } + if state.plan != operation { + return ErrStressOperationOwnerMismatch + } + if state.started { + return ErrStressOperationDuplicateStart + } + state.started = true + return nil +} + +func (registry *stressExactOnceRegistry) Complete(receipt StressOperationReceipt) error { + registry.mu.Lock() + defer registry.mu.Unlock() + state, exists := registry.operations[receipt.Operation.ID] + if !exists { + return ErrStressOperationUnknown + } + if state.plan != receipt.Operation { + return ErrStressOperationOwnerMismatch + } + if !state.started { + return ErrStressOperationDuplicateStart + } + if state.completed { + return ErrStressOperationDuplicateCompletion + } + if receipt.SuccessProof == "" { + return ErrStressOperationMissingCompletion + } + state.completed = true + return nil +} + +func (registry *stressExactOnceRegistry) ValidateComplete() error { + registry.mu.Lock() + defer registry.mu.Unlock() + for _, state := range registry.operations { + if !state.started || !state.completed { + return ErrStressOperationMissingCompletion + } + } + return nil +} diff --git a/integration/agentcompat/internal/scenario/stress_exact_once_test.go b/integration/agentcompat/internal/scenario/stress_exact_once_test.go new file mode 100644 index 00000000..bfbe0a3b --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_exact_once_test.go @@ -0,0 +1,43 @@ +//go:build linux + +package scenario + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +func TestStressExactOnceRegistryRejectsInvalidReceipts(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + registry, err := newStressExactOnceRegistry(plan) + require.NoError(t, err) + operation := plan.Rounds[0].Operations[0] + receipt := StressOperationReceipt{Operation: operation, SuccessProof: "ok"} + require.NoError(t, registry.Start(operation)) + require.ErrorIs(t, registry.Start(operation), ErrStressOperationDuplicateStart) + require.NoError(t, registry.Complete(receipt)) + require.ErrorIs(t, registry.Complete(receipt), ErrStressOperationDuplicateCompletion) + require.ErrorIs(t, registry.Start(StressOperationPlan{ID: operation.ID}), ErrStressOperationOwnerMismatch) + require.ErrorIs(t, registry.Complete(StressOperationReceipt{Operation: StressOperationPlan{ID: operation.ID, Round: 2, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind}, SuccessProof: "ok"}), ErrStressOperationOwnerMismatch) +} + +func TestStressExactOnceRegistryRequiresCanonicalCompletion(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + registry, err := newStressExactOnceRegistry(plan) + require.NoError(t, err) + for _, round := range plan.Rounds { + for _, operation := range round.Operations { + require.NoError(t, registry.Start(operation)) + } + } + require.ErrorIs(t, registry.ValidateComplete(), ErrStressOperationMissingCompletion) +} diff --git a/integration/agentcompat/internal/scenario/stress_identity.go b/integration/agentcompat/internal/scenario/stress_identity.go new file mode 100644 index 00000000..17ae8a56 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_identity.go @@ -0,0 +1,185 @@ +//go:build linux + +package scenario + +import ( + "encoding/json" + "errors" + "fmt" +) + +var ErrStressIdentity = errors.New("stress identity is invalid") + +type StressAgentOrdinal struct{ value int } + +func NewStressAgentOrdinal(value int) (StressAgentOrdinal, error) { + if value < 1 { + return StressAgentOrdinal{}, fmt.Errorf("agent ordinal %d: %w", value, ErrStressIdentity) + } + return StressAgentOrdinal{value: value}, nil +} + +func (ordinal StressAgentOrdinal) Int() int { return ordinal.value } + +func (ordinal StressAgentOrdinal) MarshalJSON() ([]byte, error) { return json.Marshal(ordinal.value) } + +func (ordinal *StressAgentOrdinal) UnmarshalJSON(data []byte) error { + var value int + if err := json.Unmarshal(data, &value); err != nil || value < 1 { + return ErrStressIdentity + } + ordinal.value = value + return nil +} + +type StressOperationID struct{ value string } + +func NewStressOperationID(value string) (StressOperationID, error) { + if value == "" { + return StressOperationID{}, ErrStressIdentity + } + return StressOperationID{value: value}, nil +} + +func (identity StressOperationID) String() string { return identity.value } + +func (identity StressOperationID) MarshalJSON() ([]byte, error) { return json.Marshal(identity.value) } + +func (identity *StressOperationID) UnmarshalJSON(data []byte) error { + var value string + if err := json.Unmarshal(data, &value); err != nil || value == "" { + return ErrStressIdentity + } + identity.value = value + return nil +} + +type StressSessionID struct{ value string } + +func NewStressSessionID(value string) (StressSessionID, error) { + if value == "" { + return StressSessionID{}, ErrStressIdentity + } + return StressSessionID{value: value}, nil +} + +func (identity StressSessionID) String() string { return identity.value } + +func (identity StressSessionID) MarshalJSON() ([]byte, error) { return json.Marshal(identity.value) } + +func (identity *StressSessionID) UnmarshalJSON(data []byte) error { + var value string + if err := json.Unmarshal(data, &value); err != nil || value == "" { + return ErrStressIdentity + } + identity.value = value + return nil +} + +type StressPATID struct{ value string } + +func NewStressPATID(value string) (StressPATID, error) { + if value == "" { + return StressPATID{}, ErrStressIdentity + } + return StressPATID{value: value}, nil +} + +func (identity StressPATID) String() string { return identity.value } + +func (identity StressPATID) MarshalJSON() ([]byte, error) { return json.Marshal(identity.value) } + +func (identity *StressPATID) UnmarshalJSON(data []byte) error { + var value string + if err := json.Unmarshal(data, &value); err != nil || value == "" { + return ErrStressIdentity + } + identity.value = value + return nil +} + +type StressOperationKind string + +const ( + StressOperationExec StressOperationKind = "exec" + StressOperationFilesystem StressOperationKind = "filesystem" +) + +type StressSessionKind string + +const ( + StressSessionTerminal StressSessionKind = "terminal" + StressSessionNAT StressSessionKind = "nat" + StressSessionFM StressSessionKind = "file-manager" +) + +type StressProcessKind string + +const ( + StressProcessDashboard StressProcessKind = "dashboard" + StressProcessAgent StressProcessKind = "agent" +) + +type StressProcessIdentity struct { + Kind StressProcessKind `json:"kind"` + Agent StressAgentOrdinal `json:"agent,omitempty"` + PID int `json:"pid"` +} + +func (identity StressProcessIdentity) MarshalJSON() ([]byte, error) { + type processWire struct { + Kind StressProcessKind `json:"kind"` + Agent *StressAgentOrdinal `json:"agent,omitempty"` + PID int `json:"pid"` + } + var agent *StressAgentOrdinal + if identity.Kind == StressProcessAgent { + if identity.Agent.Int() < 1 { + return nil, ErrStressIdentity + } + agent = &identity.Agent + } + return json.Marshal(processWire{Kind: identity.Kind, Agent: agent, PID: identity.PID}) +} + +func (identity *StressProcessIdentity) UnmarshalJSON(data []byte) error { + type processWire struct { + Kind StressProcessKind `json:"kind"` + Agent *StressAgentOrdinal `json:"agent"` + PID int `json:"pid"` + } + var wire processWire + if err := json.Unmarshal(data, &wire); err != nil || wire.PID < 1 { + return ErrStressIdentity + } + switch wire.Kind { + case StressProcessDashboard: + *identity = StressProcessIdentity{Kind: wire.Kind, PID: wire.PID} + case StressProcessAgent: + if wire.Agent == nil || wire.Agent.Int() < 1 { + return ErrStressIdentity + } + *identity = StressProcessIdentity{Kind: wire.Kind, Agent: *wire.Agent, PID: wire.PID} + default: + return ErrStressIdentity + } + return nil +} + +func NewStressDashboardProcess(pid int) (StressProcessIdentity, error) { + if pid < 1 { + return StressProcessIdentity{}, fmt.Errorf("dashboard PID %d: %w", pid, ErrStressIdentity) + } + return StressProcessIdentity{Kind: StressProcessDashboard, PID: pid}, nil +} + +func NewStressAgentProcess(agent StressAgentOrdinal, pid int) (StressProcessIdentity, error) { + if agent.Int() < 1 || pid < 1 { + return StressProcessIdentity{}, fmt.Errorf("agent process ordinal=%d PID=%d: %w", agent.Int(), pid, ErrStressIdentity) + } + return StressProcessIdentity{Kind: StressProcessAgent, Agent: agent, PID: pid}, nil +} + +func (identity StressProcessIdentity) key() string { + return fmt.Sprintf("%s:%d", identity.Kind, identity.Agent.Int()) +} diff --git a/integration/agentcompat/internal/scenario/stress_path_lock.go b/integration/agentcompat/internal/scenario/stress_path_lock.go new file mode 100644 index 00000000..d3cb41b7 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_path_lock.go @@ -0,0 +1,73 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" +) + +var ErrStressPathLockProof = errors.New("stress path-lock proof is invalid") + +type stressPathLockProof struct { + Stripes int +} + +func (proof stressPathLockProof) Validate() error { + if proof.Stripes != 1024 { + return fmt.Errorf("stripes=%d: %w", proof.Stripes, ErrStressPathLockProof) + } + return nil +} + +func proveStressPathLockStripes(ctx context.Context, sourceDir string) (stressPathLockProof, error) { + if sourceDir == "" { + return stressPathLockProof{}, ErrStressPathLockProof + } + root, err := os.MkdirTemp("", "nezha-stress-path-lock-") + if err != nil { + return stressPathLockProof{}, err + } + defer os.RemoveAll(root) + probe := filepath.Join(root, "agentcompat_path_lock_proof_test.go") + content := "package main\n\nimport (\n\t\"fmt\"\n\t\"testing\"\n)\n\nconst agentCompatPathLockStripes = fsPathLockStripes\n\nvar _ [agentCompatPathLockStripes - 1024]struct{}\nvar _ [1024 - agentCompatPathLockStripes]struct{}\n\nfunc TestAgentCompatPathLockProof(t *testing.T) {\n\tfmt.Printf(\"AGENTCOMPAT_PATH_LOCK_STRIPES=%d\\n\", agentCompatPathLockStripes)\n}\n" + if err := os.WriteFile(probe, []byte(content), 0o600); err != nil { + return stressPathLockProof{}, err + } + overlayPath := filepath.Join(root, "overlay.json") + original := filepath.Join(sourceDir, "cmd", "agent", "mcp_fs_path_lock_bounded_test.go") + overlayData, err := json.Marshal(struct { + Replace map[string]string `json:"Replace"` + }{Replace: map[string]string{original: probe}}) + if err != nil { + return stressPathLockProof{}, err + } + if err := os.WriteFile(overlayPath, overlayData, 0o600); err != nil { + return stressPathLockProof{}, err + } + command := exec.CommandContext(ctx, "go", "test", "-overlay", overlayPath, "./cmd/agent", "-run", "^TestAgentCompatPathLockProof$", "-count=1", "-v") + command.Dir = sourceDir + command.Env = append(os.Environ(), "GOFLAGS=") + output, err := command.CombinedOutput() + if err != nil { + return stressPathLockProof{}, fmt.Errorf("compile path-lock proof: %w: %s", err, output) + } + for _, line := range strings.Split(string(output), "\n") { + if !strings.HasPrefix(line, "AGENTCOMPAT_PATH_LOCK_STRIPES=") { + continue + } + value, parseErr := strconv.Atoi(strings.TrimPrefix(line, "AGENTCOMPAT_PATH_LOCK_STRIPES=")) + if parseErr == nil { + proof := stressPathLockProof{Stripes: value} + return proof, proof.Validate() + } + } + return stressPathLockProof{}, fmt.Errorf("path-lock proof output missing: %w", ErrStressPathLockProof) +} diff --git a/integration/agentcompat/internal/scenario/stress_path_lock_test.go b/integration/agentcompat/internal/scenario/stress_path_lock_test.go new file mode 100644 index 00000000..07c795c5 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_path_lock_test.go @@ -0,0 +1,35 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "crypto/sha256" + "os" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestStressPathLockProofRequiresExactStripeCount(t *testing.T) { + for _, stripes := range []int{1023, 1025} { + require.ErrorIs(t, (stressPathLockProof{Stripes: stripes}).Validate(), ErrStressPathLockProof) + } + require.NoError(t, (stressPathLockProof{Stripes: 1024}).Validate()) +} + +func TestStressPathLockProofDoesNotModifyAgentSource(t *testing.T) { + sourceDir := os.Getenv("AGENTCOMPAT_AGENT_SOURCE") + if sourceDir == "" { + t.Skip("AGENTCOMPAT_AGENT_SOURCE is not configured") + } + path := sourceDir + "/cmd/agent/mcp_fs_path_lock.go" + before, err := os.ReadFile(path) + require.NoError(t, err) + beforeHash := sha256.Sum256(before) + proof, err := proveStressPathLockStripes(t.Context(), sourceDir) + require.NoError(t, err) + require.NoError(t, proof.Validate()) + after, err := os.ReadFile(path) + require.NoError(t, err) + require.Equal(t, beforeHash, sha256.Sum256(after)) +} diff --git a/integration/agentcompat/internal/scenario/stress_plan.go b/integration/agentcompat/internal/scenario/stress_plan.go new file mode 100644 index 00000000..de06b859 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_plan.go @@ -0,0 +1,120 @@ +//go:build linux + +package scenario + +import ( + "errors" + "fmt" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +var ErrStressPlan = errors.New("stress plan is invalid") + +type StressOperationPlan struct { + ID StressOperationID `json:"id"` + Round int `json:"round"` + Agent StressAgentOrdinal `json:"agent"` + PAT StressPATID `json:"pat_id"` + Kind StressOperationKind `json:"kind"` +} + +type StressRoundPlan struct { + Round int `json:"round"` + Operations []StressOperationPlan `json:"operations"` +} + +type StressSessionPlan struct { + ID StressSessionID `json:"id"` + Kind StressSessionKind `json:"kind"` + Ordinal int `json:"ordinal"` + Agent StressAgentOrdinal `json:"agent"` +} + +type StressPlan struct { + Profile contract.ProfileName `json:"profile"` + Seed contract.Seed `json:"seed"` + Rounds []StressRoundPlan `json:"rounds"` + Sessions []StressSessionPlan `json:"sessions"` +} + +func GenerateStressPlan(profile contract.Profile, seed contract.Seed) (StressPlan, error) { + if seed == 0 || profile.AgentCount() < 1 || profile.StressRounds() < 1 || profile.ConcurrentSessions() < 1 { + return StressPlan{}, ErrStressPlan + } + expectedOperations := profile.AgentCount() * profile.StressRounds() * 2 + if profile.ConcurrentOperations() != expectedOperations { + return StressPlan{}, fmt.Errorf("concurrent operations=%d want=%d: %w", profile.ConcurrentOperations(), expectedOperations, ErrStressPlan) + } + plan := StressPlan{Profile: profile.Name(), Seed: seed} + plan.Rounds = make([]StressRoundPlan, profile.StressRounds()) + for roundIndex := range plan.Rounds { + round := roundIndex + 1 + operations, err := stressRoundOperations(seed, round, profile.AgentCount()) + if err != nil { + return StressPlan{}, err + } + plan.Rounds[roundIndex] = StressRoundPlan{Round: round, Operations: operations} + } + sessions, err := stressSessionPlans(seed, profile.AgentCount(), profile.ConcurrentSessions()) + if err != nil { + return StressPlan{}, err + } + plan.Sessions = sessions + return plan, nil +} + +func stressRoundOperations(seed contract.Seed, round, agentCount int) ([]StressOperationPlan, error) { + operations := make([]StressOperationPlan, 0, agentCount*2) + for agentValue := 1; agentValue <= agentCount; agentValue++ { + agent, err := NewStressAgentOrdinal(agentValue) + if err != nil { + return nil, err + } + pat, err := NewStressPATID(fmt.Sprintf("pat-%016x-a%02d", uint64(seed), agentValue)) + if err != nil { + return nil, err + } + for _, kind := range []StressOperationKind{StressOperationExec, StressOperationFilesystem} { + identity, idErr := NewStressOperationID(fmt.Sprintf("op-%016x-r%02d-a%02d-%s", uint64(seed), round, agentValue, kind)) + if idErr != nil { + return nil, idErr + } + operations = append(operations, StressOperationPlan{ID: identity, Round: round, Agent: agent, PAT: pat, Kind: kind}) + } + } + random := stressRandom(uint64(seed) ^ uint64(round)*0x9e3779b97f4a7c15) + for index := len(operations) - 1; index > 0; index-- { + swap := int(random.next() % uint64(index+1)) + operations[index], operations[swap] = operations[swap], operations[index] + } + return operations, nil +} + +func stressSessionPlans(seed contract.Seed, agentCount, countPerKind int) ([]StressSessionPlan, error) { + sessions := make([]StressSessionPlan, 0, countPerKind*3) + for _, kind := range []StressSessionKind{StressSessionTerminal, StressSessionNAT, StressSessionFM} { + for index := 1; index <= countPerKind; index++ { + agent, err := NewStressAgentOrdinal((index-1)%agentCount + 1) + if err != nil { + return nil, err + } + identity, err := NewStressSessionID(fmt.Sprintf("session-%016x-%s-%02d", uint64(seed), kind, index)) + if err != nil { + return nil, err + } + sessions = append(sessions, StressSessionPlan{ID: identity, Kind: kind, Ordinal: index, Agent: agent}) + } + } + return sessions, nil +} + +type stressRandom uint64 + +func (random *stressRandom) next() uint64 { + *random += 0x9e3779b97f4a7c15 + value := uint64(*random) + value = (value ^ (value >> 30)) * 0xbf58476d1ce4e5b9 + value = (value ^ (value >> 27)) * 0x94d049bb133111eb + return value ^ (value >> 31) +} diff --git a/integration/agentcompat/internal/scenario/stress_plan_test.go b/integration/agentcompat/internal/scenario/stress_plan_test.go new file mode 100644 index 00000000..06d5837a --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_plan_test.go @@ -0,0 +1,109 @@ +//go:build linux + +package scenario + +import ( + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +func TestStress_PlanHasFourRoundsSixteenOpsEach(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + + require.NoError(t, err) + require.Len(t, plan.Rounds, 4) + for _, round := range plan.Rounds { + require.Len(t, round.Operations, 16) + } +} + +func TestStress_PlanUsesTwoRequestsPerPATPerRound(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + + for _, round := range plan.Rounds { + counts := make(map[StressPATID]int) + kinds := make(map[StressPATID]map[StressOperationKind]int) + for _, operation := range round.Operations { + counts[operation.PAT]++ + if kinds[operation.PAT] == nil { + kinds[operation.PAT] = make(map[StressOperationKind]int) + } + kinds[operation.PAT][operation.Kind]++ + } + require.Len(t, counts, 8) + for pat, count := range counts { + require.Equal(t, 2, count) + require.Equal(t, 1, kinds[pat][StressOperationExec]) + require.Equal(t, 1, kinds[pat][StressOperationFilesystem]) + } + } +} + +func TestStress_PlanHas64UniqueStableIDs(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + + seen := make(map[StressOperationID]struct{}) + for _, round := range plan.Rounds { + for _, operation := range round.Operations { + _, duplicate := seen[operation.ID] + require.False(t, duplicate) + seen[operation.ID] = struct{}{} + } + } + require.Len(t, seen, 64) +} + +func TestStress_PlanFixedSeedReproducesByteForByte(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + first, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + second, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + + firstJSON, err := json.Marshal(first) + require.NoError(t, err) + secondJSON, err := json.Marshal(second) + require.NoError(t, err) + require.Equal(t, firstJSON, secondJSON) +} + +func TestStress_RejectsLaunchWindowOverOneSecond(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + evidence := successfulStressRoundEvidence(plan.Rounds[0]) + evidence.Operations[len(evidence.Operations)-1].LaunchedAt = evidence.Operations[0].LaunchedAt.Add(time.Second + time.Nanosecond) + evidence.Operations[len(evidence.Operations)-1].CompletedAt = evidence.Operations[len(evidence.Operations)-1].LaunchedAt.Add(time.Millisecond) + + err = ValidateStressRoundEvidence(plan.Rounds[0], evidence) + + require.ErrorIs(t, err, ErrStressLaunchWindow) +} + +func successfulStressRoundEvidence(plan StressRoundPlan) StressRoundEvidence { + started := time.Unix(100, 0) + operations := make([]StressOperationEvidence, len(plan.Operations)) + for index, operation := range plan.Operations { + operations[index] = StressOperationEvidence{ + ID: operation.ID, Round: operation.Round, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind, SuccessProof: "ok", + LaunchedAt: started, CompletedAt: started.Add(time.Millisecond), Succeeded: true, + } + } + return StressRoundEvidence{Round: plan.Round, Operations: operations} +} diff --git a/integration/agentcompat/internal/scenario/stress_qa_artifact_test.go b/integration/agentcompat/internal/scenario/stress_qa_artifact_test.go new file mode 100644 index 00000000..e762831f --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_qa_artifact_test.go @@ -0,0 +1,36 @@ +//go:build linux + +package scenario + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +func TestStress_WriteTypedQAArtifact(t *testing.T) { + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + fixture := stressDashboardResourceFixture(100) + evaluation, err := EvaluateStressResource(fixture) + require.NoError(t, err) + artifact := struct { + Plan StressPlan `json:"plan"` + Resource StressResourceEvaluation `json:"resource"` + }{Plan: plan, Resource: evaluation} + data, err := json.MarshalIndent(artifact, "", " ") + require.NoError(t, err) + path := filepath.Join(t.TempDir(), "stress-contracts.json") + require.NoError(t, os.WriteFile(path, data, 0o600)) + require.Contains(t, string(data), `"profile": "pr-full"`) + require.Contains(t, string(data), `"rounds"`) + require.Contains(t, string(data), `"rss_limit_bytes": 67108864`) + require.NotEmpty(t, path) +} diff --git a/integration/agentcompat/internal/scenario/stress_real_e2e_test.go b/integration/agentcompat/internal/scenario/stress_real_e2e_test.go new file mode 100644 index 00000000..d0073352 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_real_e2e_test.go @@ -0,0 +1,197 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "context" + "net/http" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/agent" + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" + "github.com/nezhahq/nezha/integration/agentcompat/internal/dashboard" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +func TestStressPRFullEightAgentExactlyOnce(t *testing.T) { + requireHeldRealSources(t) + paths, err := contract.NewPaths(os.Getenv("AGENTCOMPAT_NEZHA_SOURCE"), os.Getenv("AGENTCOMPAT_AGENT_SOURCE"), filepath.Join("/tmp", "nezha-agentcompat-real-stress")) + require.NoError(t, err) + profile, err := contract.ProfileByName(string(contract.ProfilePRFull)) + require.NoError(t, err) + plan, err := GenerateStressPlan(profile, contract.DefaultSeed) + require.NoError(t, err) + ctx, cancel := context.WithTimeout(t.Context(), contract.PRFullSuiteDeadline) + defer cancel() + realFixture, err := startHeldSessionSetRealFixture(ctx, paths, plan) + require.NoError(t, err) + t.Cleanup(func() { _ = realFixture.close(context.Background(), nil) }) + input, err := realFixture.input(plan) + require.NoError(t, err) + set, err := NewHeldSessionSet(ctx, input) + require.NoError(t, err) + dashboardIdentity := realFixture.dashboard.RuntimeIdentity() + agentIdentities := make([]agent.ProcessIdentity, len(realFixture.agents)) + workspaceRoots := make([]string, 0, len(realFixture.agents)+2) + workspaceRoots = append(workspaceRoots, realFixture.dashboard.WorkspaceRoot(), realFixture.preparedBinary.WorkspaceRoot()) + for index, instance := range realFixture.agents { + agentIdentities[index] = instance.RuntimeIdentity() + workspaceRoots = append(workspaceRoots, instance.WorkspaceRoot()) + } + t.Cleanup(func() { _ = set.Close(context.Background()) }) + warmups, warmupErr := runStressWarmups(ctx, realFixture, plan) + require.NoError(t, warmupErr) + require.NoError(t, drainStressDashboardSQLiteJournal(ctx, realFixture)) + baselineResources, resourceErr := captureStressResources(ctx, realFixture, false) + require.NoError(t, resourceErr) + rounds := make([]StressRoundEvidence, 0, len(plan.Rounds)) + for _, round := range plan.Rounds { + // WaitHealthy completes only after Close; live sessions are validated by NewHeldSessionSet. + evidenceValue, roundErr := runStressRound(ctx, realFixture, round) + require.NoError(t, roundErr) + require.NoError(t, ValidateStressRoundEvidence(round, evidenceValue)) + rounds = append(rounds, evidenceValue) + } + quota, quotaErr := runRealStressQuotaProbe(ctx, realFixture, paths.AgentSource().String()) + require.NoError(t, quotaErr) + endResources, resourceErr := captureStressResources(ctx, realFixture, true) + require.NoError(t, resourceErr) + require.NoError(t, set.Close(ctx)) + require.NoError(t, set.WaitHealthy(ctx)) + cleanupErr := realFixture.close(ctx, nil) + require.NoError(t, cleanupErr) + cleanup := stressCleanupSummary(realFixture, dashboardIdentity, agentIdentities, workspaceRoots, cleanupErr) + resourceWindows := make([]StressProcessWindows, len(baselineResources)) + for index := range baselineResources { + resourceWindows[index] = StressProcessWindows{Process: baselineResources[index].Process, Baseline: baselineResources[index].Baseline, End: endResources[index].End} + } + evidenceValue := StressEvidence{Version: 1, Profile: plan.Profile, Seed: plan.Seed, PreparedBinaries: StressPreparedBinaries{DashboardBuildCount: 1, DashboardPathReused: true, AgentBuildCount: 1, AgentPathReused: true}, Quotas: quota, Warmups: warmups, Plan: plan, Iterations: []StressIterationEvidence{{Iteration: 1, Rounds: rounds, Resources: resourceWindows}}, Cleanup: cleanup} + for _, session := range plan.Sessions { + evidenceValue.Sessions = append(evidenceValue.Sessions, StressSessionEvidence{ID: session.ID, Kind: session.Kind, Succeeded: true}) + } + require.NoError(t, publishStressEvidence("/tmp/nezha-held-real-sessions", evidenceValue)) + _, err = readStressEvidence("/tmp/nezha-held-real-sessions") + require.NoError(t, err) +} + +func captureStressResources(ctx context.Context, fixture *heldSessionSetRealFixture, end bool) ([]StressProcessWindows, error) { + result := make([]StressProcessWindows, 0, len(fixture.agents)+1) + dashboard := fixture.dashboard.RuntimeIdentity() + dashboardProcess, err := NewStressDashboardProcess(dashboard.PID) + if err != nil { + return nil, err + } + windowSpec := processharness.WindowSpec{PID: dashboard.PID, Interval: contract.ResourceSampleInterval} + if !end { + windowSpec.ObserveSample = observeStressDashboardSQLiteJournal(fixture.dashboard.DatabasePath() + "-journal") + } + baseline, err := processharness.SampleWindow(ctx, windowSpec) + if err != nil { + return nil, err + } + dashboardWindow := StressProcessWindows{Process: dashboardProcess} + if end { + dashboardWindow.End = baseline + } else { + dashboardWindow.Baseline = baseline + } + result = append(result, dashboardWindow) + for index, instance := range fixture.agents { + identity := instance.RuntimeIdentity() + agentOrdinal, err := NewStressAgentOrdinal(index + 1) + if err != nil { + return nil, err + } + process, err := NewStressAgentProcess(agentOrdinal, identity.PID) + if err != nil { + return nil, err + } + baseline, err := processharness.SampleWindow(ctx, processharness.WindowSpec{PID: identity.PID, Interval: contract.ResourceSampleInterval}) + if err != nil { + return nil, err + } + window := StressProcessWindows{Process: process} + if end { + window.End = baseline + } else { + window.Baseline = baseline + } + result = append(result, window) + } + return result, nil +} + +type realStressQuotaResponse struct { + UserAccepted int `json:"user_accepted"` + UserRejected int `json:"user_rejected"` + ServerAccepted int `json:"server_accepted"` + ServerRejected int `json:"server_rejected"` + Clean bool `json:"clean"` +} + +type realStressRateLimitResponse struct { + SecondAllowedCount int `json:"second_allowed_count"` + SecondRejectedAtCount int `json:"second_rejected_at_count"` + MinuteAllowedCount int `json:"minute_allowed_count"` + MinuteRejectedAtCount int `json:"minute_rejected_at_count"` +} + +func runRealStressQuotaProbe(ctx context.Context, fixture *heldSessionSetRealFixture, agentSource string) (StressQuotaEvidence, error) { + response, err := client.DoREST[struct{}, realStressQuotaResponse](ctx, fixture.controlPAT.Client, client.RESTRequest[struct{}]{Method: http.MethodPost, Path: "/agentcompat/io-stream-quota-probe", Body: &struct{}{}}) + if err != nil { + return StressQuotaEvidence{}, err + } + rateLimit, err := client.DoREST[struct{}, realStressRateLimitResponse](ctx, fixture.controlPAT.Client, client.RESTRequest[struct{}]{Method: http.MethodPost, Path: "/agentcompat/mcp-rate-limit-probe", Body: &struct{}{}}) + if err != nil { + return StressQuotaEvidence{}, err + } + pathLockProof, err := proveStressPathLockStripes(ctx, agentSource) + if err != nil { + return StressQuotaEvidence{}, err + } + return StressQuotaEvidence{PATSecond: StressQuotaBoundary{Allowed: rateLimit.SecondAllowedCount, Rejected: rateLimit.SecondRejectedAtCount, AllowedAccepted: rateLimit.SecondAllowedCount == 10, RejectedDenied: rateLimit.SecondRejectedAtCount == 11}, PATMinute: StressQuotaBoundary{Allowed: rateLimit.MinuteAllowedCount, Rejected: rateLimit.MinuteRejectedAtCount, AllowedAccepted: rateLimit.MinuteAllowedCount == 120, RejectedDenied: rateLimit.MinuteRejectedAtCount == 121}, UserStreams: StressQuotaBoundary{Allowed: response.UserAccepted, Rejected: response.UserAccepted + response.UserRejected, AllowedAccepted: response.UserAccepted == 20, RejectedDenied: response.UserRejected == 1}, ServerStreams: StressQuotaBoundary{Allowed: response.ServerAccepted, Rejected: response.ServerAccepted + response.ServerRejected, AllowedAccepted: response.ServerAccepted == 40, RejectedDenied: response.ServerRejected == 1}, PathLockStripes: pathLockProof.Stripes}, nil +} + +func stressCleanupSummary(fixture *heldSessionSetRealFixture, dashboardIdentity dashboard.RuntimeIdentity, agentIdentities []agent.ProcessIdentity, workspaceRoots []string, cleanupErr error) StressCleanupSummary { + summary := StressCleanupSummary{ReceiptCount: 1 + len(fixture.agents), WorkspaceResidue: 0} + receipts := make([]processharness.CleanupReceipt, 0, summary.ReceiptCount) + receipts = append(receipts, fixture.dashboard.CleanupReceipt()) + for _, instance := range fixture.agents { + receipts = append(receipts, instance.CleanupReceipt()) + } + for _, receipt := range receipts { + if !receipt.Passed { + summary.FailedReceiptCount++ + } + if receipt.Forced { + summary.ForcedCleanupCount++ + } + } + if !heldRealPIDGone(dashboardIdentity.PID) { + summary.ProcessResidue++ + } + if !heldRealGroupGone(dashboardIdentity.ProcessGroupID) { + summary.ProcessGroupResidue++ + } + for index := range fixture.agents { + identity := agentIdentities[index] + if !heldRealPIDGone(identity.PID) { + summary.ProcessResidue++ + } + if !heldRealGroupGone(identity.ProcessGroupID) { + summary.ProcessGroupResidue++ + } + } + for _, root := range workspaceRoots { + if !heldSessionSetRealWorkspaceGone(root) { + summary.WorkspaceResidue++ + } + } + summary.Passed = cleanupErr == nil && summary.ReceiptCount == 9 && summary.FailedReceiptCount == 0 && summary.ForcedCleanupCount == 0 && summary.ProcessResidue == 0 && summary.ProcessGroupResidue == 0 && summary.WorkspaceResidue == 0 + return summary +} diff --git a/integration/agentcompat/internal/scenario/stress_real_operations.go b/integration/agentcompat/internal/scenario/stress_real_operations.go new file mode 100644 index 00000000..f9e89928 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_real_operations.go @@ -0,0 +1,152 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "os" + "time" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + "github.com/nezhahq/nezha/integration/agentcompat/internal/fixture" +) + +type stressOperationExecutor struct { + fixture *heldSessionSetRealFixture + plan StressOperationPlan +} + +func (executor stressOperationExecutor) run(ctx context.Context) StressOperationEvidence { + started := time.Now() + proof, err := executor.execute(ctx) + completed := time.Now() + evidenceValue := StressOperationEvidence{ID: executor.plan.ID, Round: executor.plan.Round, Agent: executor.plan.Agent, PAT: executor.plan.PAT, Kind: executor.plan.Kind, LaunchedAt: started, CompletedAt: completed, Succeeded: err == nil, SuccessProof: proof} + if err != nil { + evidenceValue.Error = errorText(err) + } + return evidenceValue +} + +func (executor stressOperationExecutor) execute(ctx context.Context) (string, error) { + if executor.fixture == nil || executor.plan.Agent.Int() < 1 || executor.plan.Agent.Int() > len(executor.fixture.agents) || executor.plan.Agent.Int() > len(executor.fixture.readiness) || executor.plan.Agent.Int() > len(executor.fixture.agentPATs) { + return "", errors.New("stress operation fixture mapping is invalid") + } + if executor.plan.PAT.String() == "" { + return "", errors.New("stress operation PAT is empty") + } + serverID := executor.fixture.readiness[executor.plan.Agent.Int()-1].ServerID + patIdentity, err := stressOperationPATIdentity(executor.fixture, executor.plan) + if err != nil { + return "", err + } + switch executor.plan.Kind { + case StressOperationExec: + return executor.exec(ctx, patIdentity.Client, serverID) + case StressOperationFilesystem: + return executor.filesystem(ctx, patIdentity.Client, serverID) + default: + return "", errors.New("unsupported stress operation kind") + } +} + +func stressOperationPATIdentity(fixture *heldSessionSetRealFixture, plan StressOperationPlan) (heldRealPATIdentity, error) { + index := plan.Agent.Int() - 1 + for _, round := range fixture.plan.Rounds { + for _, planned := range round.Operations { + if planned.ID == plan.ID { + if planned.Agent != plan.Agent || planned.Kind != plan.Kind || planned.PAT != plan.PAT { + return heldRealPATIdentity{}, errors.New("stress operation plan PAT mapping is invalid") + } + return fixture.agentPATs[index], nil + } + } + } + for _, round := range fixture.plan.Rounds { + for _, planned := range round.Operations { + if planned.Agent == plan.Agent && planned.PAT == plan.PAT && planned.Kind == plan.Kind { + return fixture.agentPATs[index], nil + } + } + } + return heldRealPATIdentity{}, errors.New("stress operation is absent from canonical plan") +} + +func runStressWarmups(ctx context.Context, fixture *heldSessionSetRealFixture, plan StressPlan) ([]StressWarmupEvidence, error) { + warmups := make([]StressWarmupEvidence, 0, len(fixture.agents)) + for agentIndex := range fixture.agents { + agentOrdinal, err := NewStressAgentOrdinal(agentIndex + 1) + if err != nil { + return nil, err + } + pat := StressPATID{} + for _, operation := range plan.Rounds[0].Operations { + if operation.Agent == agentOrdinal { + pat = operation.PAT + break + } + } + if pat.String() == "" { + return nil, errors.New("stress warmup PAT mapping is invalid") + } + execID, err := NewStressOperationID(fmt.Sprintf("warmup-exec-a%02d", agentOrdinal.Int())) + if err != nil { + return nil, err + } + execResult := stressOperationExecutor{fixture: fixture, plan: StressOperationPlan{ID: execID, Agent: agentOrdinal, PAT: pat, Kind: StressOperationExec}} + if _, err := execResult.execute(ctx); err != nil { + return nil, err + } + filesystemID, err := NewStressOperationID(fmt.Sprintf("warmup-filesystem-a%02d", agentOrdinal.Int())) + if err != nil { + return nil, err + } + filesystemResult := stressOperationExecutor{fixture: fixture, plan: StressOperationPlan{ID: filesystemID, Round: 0, Agent: agentOrdinal, PAT: pat, Kind: StressOperationFilesystem}} + if _, err := filesystemResult.execute(ctx); err != nil { + return nil, err + } + warmups = append(warmups, StressWarmupEvidence{Agent: agentOrdinal, Exec: true, Filesystem: true, Terminal: true, NAT: true, FM: true}) + } + return warmups, nil +} + +func (executor stressOperationExecutor) exec(ctx context.Context, patClient *client.Client, serverID uint64) (string, error) { + const token = "agentcompat-stress-exec-proof" + result, err := client.CallTool[execArguments, execResult](ctx, patClient, client.ToolCall[execArguments]{Name: "server.exec", Arguments: execArguments{ServerID: serverID, Cmd: "/bin/sh", Args: []string{"-c", "printf " + token}}}) + if err != nil || result.StructuredContent.ExitCode != 0 || result.StructuredContent.Stdout != token || result.StructuredContent.Error != "" || result.StructuredContent.TimedOut || result.StructuredContent.StdoutTruncated { + return "", errors.New("stress Exec proof failed") + } + return stressProof(token), nil +} + +func (executor stressOperationExecutor) filesystem(ctx context.Context, patClient *client.Client, serverID uint64) (string, error) { + parent := executor.fixture.agents[executor.plan.Agent.Int()-1].WorkspaceRoot() + return executeStressFilesystemProof(ctx, patClient, serverID, parent, executor.plan.ID.String(), executor.plan.Round) +} + +func executeStressFilesystemProof(ctx context.Context, patClient *client.Client, serverID uint64, parent, operationID string, round int) (string, error) { + root, err := fixture.NewAgentRoot(parent, fmt.Sprintf("stress-%s", operationID)) + if err != nil { + return "", err + } + defer os.RemoveAll(root.Absolute()) + filesystem := newMCPFilesystemClient(patClient, serverID, root) + content := "agentcompat-stress-filesystem-proof" + relative := fmt.Sprintf("round-%d/%s.txt", round, operationID) + written, err := filesystem.write(ctx, mcpFilesystemWrite{relative: relative, content: content, encoding: "utf8", mode: "0600", createDirs: true}) + if err != nil { + return "", fmt.Errorf("stress filesystem write proof failed: %w", err) + } + if written.StructuredContent.Size != int64(len(content)) || written.StructuredContent.SHA256 != stressProof(content) || written.StructuredContent.Error != "" { + return "", errors.New("stress filesystem write proof response invalid") + } + return stressProof(content), nil +} + +func stressProof(value string) string { + digest := sha256.Sum256([]byte(value)) + return hex.EncodeToString(digest[:]) +} diff --git a/integration/agentcompat/internal/scenario/stress_real_operations_test.go b/integration/agentcompat/internal/scenario/stress_real_operations_test.go new file mode 100644 index 00000000..02c97553 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_real_operations_test.go @@ -0,0 +1,78 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" +) + +func TestStressFilesystemUsesAgentPAT(t *testing.T) { + fixture := &heldSessionSetRealFixture{ + agentPATs: []heldRealPATIdentity{{TokenID: 1}}, + } + operation := StressOperationPlan{Agent: mustStressAgentOrdinal(t, 1), PAT: mustStressPATID(t, "pat-1"), Kind: StressOperationFilesystem} + + fixture.plan = StressPlan{Rounds: []StressRoundPlan{{Operations: []StressOperationPlan{operation}}}} + selected, err := stressOperationPATIdentity(fixture, operation) + require.NoError(t, err) + require.Equal(t, uint64(1), selected.TokenID) +} + +func TestStressExecUsesAgentPAT(t *testing.T) { + fixture := &heldSessionSetRealFixture{ + agentPATs: []heldRealPATIdentity{{TokenID: 1}}, + } + operation := StressOperationPlan{Agent: mustStressAgentOrdinal(t, 1), PAT: mustStressPATID(t, "pat-1"), Kind: StressOperationExec} + + fixture.plan = StressPlan{Rounds: []StressRoundPlan{{Operations: []StressOperationPlan{operation}}}} + selected, err := stressOperationPATIdentity(fixture, operation) + require.NoError(t, err) + require.Equal(t, uint64(1), selected.TokenID) +} + +func TestStressFilesystemProofDispatchesOneWrite(t *testing.T) { + requests := make([]string, 0, 1) + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + var envelope struct { + Params struct { + Name string `json:"name"` + } `json:"params"` + } + require.NoError(t, json.NewDecoder(request.Body).Decode(&envelope)) + requests = append(requests, envelope.Params.Name) + writer.Header().Set("Content-Type", "application/json") + _, err := writer.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"content":[],"structuredContent":{"size":35,"sha256":"6ececdd71257073948afc9c699d12d3075d05d3dc339c4511496c3fbc27a2081"}}}`)) + require.NoError(t, err) + })) + t.Cleanup(server.Close) + mcpClient, err := client.New(client.Config{BaseURL: server.URL}) + require.NoError(t, err) + parent := t.TempDir() + + proof, err := executeStressFilesystemProof(t.Context(), mcpClient, 17, parent, "operation", 1) + + require.NoError(t, err) + require.Equal(t, stressProof("agentcompat-stress-filesystem-proof"), proof) + require.Equal(t, []string{"fs.write"}, requests) +} + +func mustStressAgentOrdinal(t *testing.T, value int) StressAgentOrdinal { + t.Helper() + ordinal, err := NewStressAgentOrdinal(value) + require.NoError(t, err) + return ordinal +} + +func mustStressPATID(t *testing.T, value string) StressPATID { + t.Helper() + pat, err := NewStressPATID(value) + require.NoError(t, err) + return pat +} diff --git a/integration/agentcompat/internal/scenario/stress_resource.go b/integration/agentcompat/internal/scenario/stress_resource.go new file mode 100644 index 00000000..fced09b7 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_resource.go @@ -0,0 +1,108 @@ +//go:build linux + +package scenario + +import ( + "errors" + "fmt" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +var ( + ErrStressProcessWindow = errors.New("stress process window is invalid") + ErrStressResourceDrift = errors.New("stress process resource count drifted") + ErrStressRSSLimit = errors.New("stress process RSS limit exceeded") +) + +type StressProcessWindows struct { + Process StressProcessIdentity `json:"process"` + Baseline processharness.Window `json:"baseline"` + End processharness.Window `json:"end"` +} + +type StressResourceMaxima struct { + RSSBytes uint64 `json:"rss_bytes"` + Descendants int `json:"descendants"` + NonStdioFDs int `json:"non_stdio_fds"` + TCPListeners int `json:"tcp_listeners"` + TCP6Listeners int `json:"tcp6_listeners"` +} + +type StressResourceEvaluation struct { + Process StressProcessIdentity `json:"process"` + Baseline StressResourceMaxima `json:"baseline"` + End StressResourceMaxima `json:"end"` + RSSDeltaBytes uint64 `json:"rss_delta_bytes"` + RSSLimitBytes uint64 `json:"rss_limit_bytes"` +} + +func EvaluateStressResource(input StressProcessWindows) (StressResourceEvaluation, error) { + baseline, err := stressWindowMaxima(input.Process, input.Baseline) + if err != nil { + return StressResourceEvaluation{}, err + } + end, err := stressWindowMaxima(input.Process, input.End) + if err != nil { + return StressResourceEvaluation{}, err + } + budget := contract.DefaultResourceBudget() + if end.Descendants != baseline.Descendants || end.NonStdioFDs != baseline.NonStdioFDs || end.TCPListeners != baseline.TCPListeners || end.TCP6Listeners != baseline.TCP6Listeners { + return StressResourceEvaluation{}, fmt.Errorf("process=%s baseline=%+v end=%+v baseline_samples=%+v end_samples=%+v: %w", input.Process.key(), baseline, end, resourceCountSamples(input.Baseline), resourceCountSamples(input.End), ErrStressResourceDrift) + } + limit, err := stressRSSLimit(input.Process, budget) + if err != nil { + return StressResourceEvaluation{}, err + } + delta := uint64(0) + if end.RSSBytes > baseline.RSSBytes { + delta = end.RSSBytes - baseline.RSSBytes + } + if delta > limit { + return StressResourceEvaluation{}, fmt.Errorf("process=%s RSS delta=%d limit=%d: %w", input.Process.key(), delta, limit, ErrStressRSSLimit) + } + return StressResourceEvaluation{Process: input.Process, Baseline: baseline, End: end, RSSDeltaBytes: delta, RSSLimitBytes: limit}, nil +} + +func resourceCountSamples(window processharness.Window) []string { + result := make([]string, 0, len(window.Samples)) + for _, sample := range window.Samples { + result = append(result, fmt.Sprintf("descendants=%d fd=%d tcp=%d tcp6=%d", sample.DescendantCount, sample.NonStdioFDCount, sample.TCPListenerCount, sample.TCP6ListenerCount)) + } + return result +} + +func stressWindowMaxima(process StressProcessIdentity, window processharness.Window) (StressResourceMaxima, error) { + if process.PID < 1 || window.PID != process.PID || len(window.Samples) != contract.ResourceSampleCount { + return StressResourceMaxima{}, fmt.Errorf("process=%s window PID=%d samples=%d: %w", process.key(), window.PID, len(window.Samples), ErrStressProcessWindow) + } + maxima := StressResourceMaxima{} + for _, sample := range window.Samples { + if sample.PID != process.PID || sample.PID < 1 { + return StressResourceMaxima{}, fmt.Errorf("process=%s sample PID=%d: %w", process.key(), sample.PID, ErrStressProcessWindow) + } + maxima.RSSBytes = max(maxima.RSSBytes, sample.RSSBytes) + // Count drift compares each window's terminal state. Using a baseline high-water + // mark turns legitimate short-lived work that has already drained into a leak. + maxima.Descendants = sample.DescendantCount + maxima.NonStdioFDs = sample.NonStdioFDCount + maxima.TCPListeners = sample.TCPListenerCount + maxima.TCP6Listeners = sample.TCP6ListenerCount + } + return maxima, nil +} + +func stressRSSLimit(process StressProcessIdentity, budget contract.ResourceBudget) (uint64, error) { + switch process.Kind { + case StressProcessDashboard: + return budget.DashboardRSSDeltaBytes(), nil + case StressProcessAgent: + if process.Agent.Int() < 1 { + return 0, fmt.Errorf("process=%s: %w", process.key(), ErrStressIdentity) + } + return budget.AgentRSSDeltaBytes(), nil + default: + return 0, fmt.Errorf("process kind=%q: %w", process.Kind, ErrStressIdentity) + } +} diff --git a/integration/agentcompat/internal/scenario/stress_resource_test.go b/integration/agentcompat/internal/scenario/stress_resource_test.go new file mode 100644 index 00000000..03e3c105 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_resource_test.go @@ -0,0 +1,178 @@ +//go:build linux + +package scenario + +import ( + "testing" + + "github.com/stretchr/testify/require" + + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +func TestStress_RejectsLeakedFD(t *testing.T) { + input := stressDashboardResourceFixture(100) + for index := range input.End.Samples { + input.End.Samples[index].NonStdioFDCount++ + } + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsSingleSampleFDTransient(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.End.Samples[4].NonStdioFDCount++ + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_AcceptsRecoveredBaselineFDTransient(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.Baseline.Samples[1].NonStdioFDCount = 9 + + evaluation, err := EvaluateStressResource(input) + + require.NoError(t, err) + require.Equal(t, 3, evaluation.Baseline.NonStdioFDs) + require.Equal(t, 3, evaluation.End.NonStdioFDs) +} + +func TestStress_RejectsSingleSampleDescendantTransient(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.End.Samples[4].DescendantCount++ + _, err := EvaluateStressResource(input) + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsSingleSampleTCPTransient(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.End.Samples[4].TCPListenerCount++ + _, err := EvaluateStressResource(input) + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsSingleSampleTCP6Transient(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.End.Samples[4].TCP6ListenerCount++ + _, err := EvaluateStressResource(input) + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsDecreasedDescendants(t *testing.T) { + input := stressDashboardResourceFixture(100) + for index := range input.End.Samples { + input.End.Samples[index].DescendantCount = 0 + input.Baseline.Samples[index].DescendantCount = 1 + } + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsDecreasedFDs(t *testing.T) { + input := stressDashboardResourceFixture(100) + for index := range input.End.Samples { + input.End.Samples[index].NonStdioFDCount = 2 + input.Baseline.Samples[index].NonStdioFDCount = 3 + } + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsDecreasedTCPListeners(t *testing.T) { + input := stressDashboardResourceFixture(100) + for index := range input.End.Samples { + input.End.Samples[index].TCPListenerCount = 0 + input.Baseline.Samples[index].TCPListenerCount = 1 + } + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsDecreasedTCP6Listeners(t *testing.T) { + input := stressDashboardResourceFixture(100) + for index := range input.End.Samples { + input.End.Samples[index].TCP6ListenerCount = 0 + input.Baseline.Samples[index].TCP6ListenerCount = 1 + } + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressResourceDrift) +} + +func TestStress_RejectsDashboardRSSOverLimit(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.End.Samples[4].RSSBytes = 100 + 67108865 + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressRSSLimit) +} + +func TestStress_RejectsAgentRSSOverLimit(t *testing.T) { + agent, err := NewStressAgentOrdinal(4) + require.NoError(t, err) + identity, err := NewStressAgentProcess(agent, 404) + require.NoError(t, err) + input := StressProcessWindows{Process: identity, Baseline: stressWindow(404, 100), End: stressWindow(404, 100+33554433)} + + _, err = EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressRSSLimit) +} + +func TestStress_RejectsVanishedPID(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.End.Samples[2].PID = 0 + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressProcessWindow) +} + +func TestStress_RejectsIncompleteSampleWindow(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.End.Samples = input.End.Samples[:4] + + _, err := EvaluateStressResource(input) + + require.ErrorIs(t, err, ErrStressProcessWindow) +} + +func TestStress_UsesFinalCountSampleButWindowRSSMaximum(t *testing.T) { + input := stressDashboardResourceFixture(100) + input.Baseline.Samples[0].NonStdioFDCount = 4 + input.End.Samples[0].NonStdioFDCount = 4 + input.End.Samples[4].RSSBytes = 110 + + evaluation, err := EvaluateStressResource(input) + + require.NoError(t, err) + require.Equal(t, uint64(10), evaluation.RSSDeltaBytes) +} + +func stressDashboardResourceFixture(rss uint64) StressProcessWindows { + identity, err := NewStressDashboardProcess(101) + if err != nil { + panic(err) + } + return StressProcessWindows{Process: identity, Baseline: stressWindow(101, rss), End: stressWindow(101, rss)} +} + +func stressWindow(pid int, rss uint64) processharness.Window { + samples := make([]processharness.Sample, 5) + for index := range samples { + samples[index] = processharness.Sample{PID: pid, RSSBytes: rss, NonStdioFDCount: 3, TCPListenerCount: 1} + } + return processharness.Window{PID: pid, Samples: samples} +} diff --git a/integration/agentcompat/internal/scenario/stress_round.go b/integration/agentcompat/internal/scenario/stress_round.go new file mode 100644 index 00000000..1e3ffe08 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_round.go @@ -0,0 +1,90 @@ +//go:build linux + +package scenario + +import ( + "errors" + "fmt" + "time" +) + +var ( + ErrStressOperationSet = errors.New("stress operation evidence set is invalid") + ErrStressLaunchWindow = errors.New("stress operation launch window exceeded one second") +) + +const stressLaunchWindow = time.Second + +type StressOperationEvidence struct { + ID StressOperationID `json:"id"` + Round int `json:"round"` + Agent StressAgentOrdinal `json:"agent"` + PAT StressPATID `json:"pat_id"` + Kind StressOperationKind `json:"kind"` + LaunchedAt time.Time `json:"launched_at"` + CompletedAt time.Time `json:"completed_at"` + Succeeded bool `json:"succeeded"` + SuccessProof string `json:"success_proof"` + Error string `json:"error,omitempty"` +} + +type StressRoundEvidence struct { + Round int `json:"round"` + Operations []StressOperationEvidence `json:"operations"` +} + +func ValidateStressRoundEvidence(plan StressRoundPlan, evidence StressRoundEvidence) error { + matched, err := matchStressRoundEvidence(plan, evidence) + if err != nil { + return err + } + for _, operation := range matched { + if !operation.Succeeded || operation.SuccessProof == "" || operation.Error != "" { + return fmt.Errorf("operation %s did not succeed: %w", operation.ID.String(), ErrStressOperationSet) + } + } + return nil +} + +func matchStressRoundEvidence(plan StressRoundPlan, evidence StressRoundEvidence) ([]StressOperationEvidence, error) { + if evidence.Round != plan.Round || len(evidence.Operations) != len(plan.Operations) { + return nil, fmt.Errorf("round=%d operations=%d want round=%d operations=%d: %w", evidence.Round, len(evidence.Operations), plan.Round, len(plan.Operations), ErrStressOperationSet) + } + expected := make(map[StressOperationID]struct{}, len(plan.Operations)) + for _, operation := range plan.Operations { + expected[operation.ID] = struct{}{} + } + matched := make([]StressOperationEvidence, 0, len(evidence.Operations)) + seen := make(map[StressOperationID]struct{}, len(evidence.Operations)) + var firstLaunch, lastLaunch time.Time + for index, operation := range evidence.Operations { + planned := plan.Operations[index] + if _, exists := expected[operation.ID]; !exists { + return nil, fmt.Errorf("unexpected operation %s: %w", operation.ID.String(), ErrStressOperationSet) + } + if operation.Round != planned.Round || operation.Agent != planned.Agent || operation.PAT != planned.PAT || operation.Kind != planned.Kind || operation.ID != planned.ID { + return nil, fmt.Errorf("operation %s owner or order mismatch: %w", operation.ID.String(), ErrStressOperationSet) + } + if _, duplicate := seen[operation.ID]; duplicate { + return nil, fmt.Errorf("duplicate operation %s: %w", operation.ID.String(), ErrStressOperationSet) + } + if operation.LaunchedAt.IsZero() || operation.CompletedAt.Before(operation.LaunchedAt) { + return nil, fmt.Errorf("operation %s timing is invalid: %w", operation.ID.String(), ErrStressOperationSet) + } + if operation.Succeeded && operation.SuccessProof == "" { + return nil, fmt.Errorf("operation %s proof is empty: %w", operation.ID.String(), ErrStressOperationSet) + } + seen[operation.ID] = struct{}{} + matched = append(matched, operation) + if firstLaunch.IsZero() || operation.LaunchedAt.Before(firstLaunch) { + firstLaunch = operation.LaunchedAt + } + if lastLaunch.IsZero() || operation.LaunchedAt.After(lastLaunch) { + lastLaunch = operation.LaunchedAt + } + } + if lastLaunch.Sub(firstLaunch) > stressLaunchWindow { + return nil, fmt.Errorf("launch window=%s: %w", lastLaunch.Sub(firstLaunch), ErrStressLaunchWindow) + } + return matched, nil +} diff --git a/integration/agentcompat/internal/scenario/stress_round_run.go b/integration/agentcompat/internal/scenario/stress_round_run.go new file mode 100644 index 00000000..fb5df7ed --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_round_run.go @@ -0,0 +1,91 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "context" + "errors" + "fmt" + "sync" +) + +func runStressRound(ctx context.Context, fixture *heldSessionSetRealFixture, plan StressRoundPlan) (StressRoundEvidence, error) { + registry, err := newStressExactOnceRegistry(StressPlan{Rounds: []StressRoundPlan{plan}}) + if err != nil { + return StressRoundEvidence{}, err + } + ready := make(chan struct{}, len(plan.Operations)) + release := make(chan struct{}) + results := make(chan StressOperationEvidence, len(plan.Operations)) + var workers sync.WaitGroup + for _, operation := range plan.Operations { + operation := operation + workers.Add(1) + go func() { + defer workers.Done() + ready <- struct{}{} + <-release + if startErr := registry.Start(operation); startErr != nil { + results <- StressOperationEvidence{ID: operation.ID, Round: operation.Round, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind, Error: startErr.Error()} + return + } + results <- stressOperationExecutor{fixture: fixture, plan: operation}.run(ctx) + }() + } + for range plan.Operations { + select { + case <-ready: + case <-ctx.Done(): + close(release) + workers.Wait() + return StressRoundEvidence{}, ctx.Err() + } + } + close(release) + workers.Wait() + byID := make(map[StressOperationID]StressOperationEvidence, len(plan.Operations)) + for range plan.Operations { + select { + case operation := <-results: + if _, duplicate := byID[operation.ID]; duplicate { + return StressRoundEvidence{}, fmt.Errorf("duplicate result for operation %s", operation.ID.String()) + } + byID[operation.ID] = operation + case <-ctx.Done(): + return StressRoundEvidence{}, ctx.Err() + } + } + evidenceValue := StressRoundEvidence{Round: plan.Round, Operations: make([]StressOperationEvidence, len(plan.Operations))} + for index, operation := range plan.Operations { + result, exists := byID[operation.ID] + if !exists { + return StressRoundEvidence{}, ErrStressOperationMissingCompletion + } + evidenceValue.Operations[index] = result + if result.Error != "" { + return StressRoundEvidence{}, errors.Join(fmt.Errorf("operation %s (%s) failed: %s", result.ID.String(), result.Kind, result.Error), stressRoundErrors(evidenceValue)) + } + if result.Error == "" { + if err := registry.Complete(StressOperationReceipt{Operation: operation, SuccessProof: result.SuccessProof}); err != nil { + return StressRoundEvidence{}, err + } + } + } + if err := registry.ValidateComplete(); err != nil { + return StressRoundEvidence{}, err + } + if err := ValidateStressRoundEvidence(plan, evidenceValue); err != nil { + return StressRoundEvidence{}, errors.Join(err, stressRoundErrors(evidenceValue)) + } + return evidenceValue, nil +} + +func stressRoundErrors(evidenceValue StressRoundEvidence) error { + var joined error + for _, operation := range evidenceValue.Operations { + if operation.Error != "" { + joined = errors.Join(joined, errors.New(operation.Error)) + } + } + return joined +} diff --git a/integration/agentcompat/internal/scenario/stress_soak.go b/integration/agentcompat/internal/scenario/stress_soak.go new file mode 100644 index 00000000..3deae952 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_soak.go @@ -0,0 +1,88 @@ +//go:build linux + +package scenario + +import ( + "errors" + "fmt" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +var ErrStressSoakTrend = errors.New("stress soak RSS increases strictly across all iterations") + +type StressRSSSeries struct { + Process StressProcessIdentity `json:"process"` + EndRSSBytes [3]uint64 `json:"end_rss_bytes"` +} + +type StressSoakTrendEvidence struct { + Series []StressRSSSeries `json:"series"` +} + +func ValidateStressSoakTrend(evidence StressSoakTrendEvidence) error { + if len(evidence.Series) == 0 { + return errors.New("stress soak RSS series are empty") + } + seen := make(map[string]struct{}, len(evidence.Series)) + for _, series := range evidence.Series { + key := series.Process.key() + if series.Process.PID < 1 || key == ":0" || (series.Process.Kind == StressProcessAgent && series.Process.Agent.Int() < 1) { + return fmt.Errorf("process=%s: %w", key, ErrStressIdentity) + } + if _, duplicate := seen[key]; duplicate { + return fmt.Errorf("duplicate process=%s: %w", key, ErrStressIdentity) + } + seen[key] = struct{}{} + values := series.EndRSSBytes + if values[0] < values[1] && values[1] < values[2] { + return fmt.Errorf("process=%s RSS=%v: %w", key, values, ErrStressSoakTrend) + } + } + return nil +} + +func ValidateStressSoakTrendForProfile(profile contract.Profile, evidence StressSoakTrendEvidence) error { + want := profile.AgentCount() + 1 + if len(evidence.Series) != want { + return fmt.Errorf("soak series=%d want=%d: %w", len(evidence.Series), want, ErrStressIdentity) + } + seenDashboard := 0 + seenAgents := make(map[int]struct{}, profile.AgentCount()) + for _, series := range evidence.Series { + if err := validateStressSoakSeries(series); err != nil { + return err + } + switch series.Process.Kind { + case StressProcessDashboard: + seenDashboard++ + case StressProcessAgent: + ordinal := series.Process.Agent.Int() + if ordinal < 1 || ordinal > profile.AgentCount() { + return fmt.Errorf("unknown agent ordinal=%d: %w", ordinal, ErrStressIdentity) + } + if _, duplicate := seenAgents[ordinal]; duplicate { + return fmt.Errorf("duplicate agent ordinal=%d: %w", ordinal, ErrStressIdentity) + } + seenAgents[ordinal] = struct{}{} + default: + return fmt.Errorf("unknown process kind=%q: %w", series.Process.Kind, ErrStressIdentity) + } + } + if seenDashboard != 1 || len(seenAgents) != profile.AgentCount() { + return fmt.Errorf("dashboard=%d agents=%d: %w", seenDashboard, len(seenAgents), ErrStressIdentity) + } + return nil +} + +func validateStressSoakSeries(series StressRSSSeries) error { + key := series.Process.key() + if series.Process.PID < 1 || key == ":0" || (series.Process.Kind == StressProcessAgent && series.Process.Agent.Int() < 1) { + return fmt.Errorf("process=%s: %w", key, ErrStressIdentity) + } + values := series.EndRSSBytes + if values[0] < values[1] && values[1] < values[2] { + return fmt.Errorf("process=%s RSS=%v: %w", key, values, ErrStressSoakTrend) + } + return nil +} diff --git a/integration/agentcompat/internal/scenario/stress_soak_test.go b/integration/agentcompat/internal/scenario/stress_soak_test.go new file mode 100644 index 00000000..dde395ee --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_soak_test.go @@ -0,0 +1,86 @@ +//go:build linux + +package scenario + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/contract" +) + +func TestStress_RejectsStrictThreePointDashboardRSSIncrease(t *testing.T) { + identity, err := NewStressDashboardProcess(101) + require.NoError(t, err) + trend := StressSoakTrendEvidence{Series: []StressRSSSeries{{Process: identity, EndRSSBytes: [3]uint64{100, 101, 102}}}} + + err = ValidateStressSoakTrend(trend) + + require.ErrorIs(t, err, ErrStressSoakTrend) +} + +func TestStress_RejectsStrictThreePointAgentRSSIncrease(t *testing.T) { + agent, err := NewStressAgentOrdinal(4) + require.NoError(t, err) + identity, err := NewStressAgentProcess(agent, 404) + require.NoError(t, err) + trend := StressSoakTrendEvidence{Series: []StressRSSSeries{{Process: identity, EndRSSBytes: [3]uint64{200, 300, 400}}}} + + err = ValidateStressSoakTrend(trend) + + require.ErrorIs(t, err, ErrStressSoakTrend) +} + +func TestStress_AcceptsStableAndNonMonotonicSoakTrend(t *testing.T) { + dashboard, err := NewStressDashboardProcess(101) + require.NoError(t, err) + agentOrdinal, err := NewStressAgentOrdinal(1) + require.NoError(t, err) + agent, err := NewStressAgentProcess(agentOrdinal, 201) + require.NoError(t, err) + trend := StressSoakTrendEvidence{Series: []StressRSSSeries{ + {Process: dashboard, EndRSSBytes: [3]uint64{100, 100, 100}}, + {Process: agent, EndRSSBytes: [3]uint64{200, 220, 210}}, + }} + + err = ValidateStressSoakTrend(trend) + + require.NoError(t, err) +} + +func TestStressSoak_RejectsMissingUnknownAndDuplicateSeries(t *testing.T) { + profile := mustProfile(t, contract.ProfileSoak) + dashboard, err := NewStressDashboardProcess(101) + require.NoError(t, err) + series := StressRSSSeries{Process: dashboard, EndRSSBytes: [3]uint64{100, 100, 100}} + + for _, mutate := range []func([]StressRSSSeries) []StressRSSSeries{ + func(values []StressRSSSeries) []StressRSSSeries { return values[:1] }, + func(values []StressRSSSeries) []StressRSSSeries { + agent, _ := NewStressAgentOrdinal(999) + process, _ := NewStressAgentProcess(agent, 999) + return append(values, StressRSSSeries{Process: process}) + }, + func(values []StressRSSSeries) []StressRSSSeries { return append(values, values[0]) }, + } { + values := make([]StressRSSSeries, 0, profile.AgentCount()+1) + values = append(values, series) + for index := 1; index <= profile.AgentCount(); index++ { + agent, agentErr := NewStressAgentOrdinal(index) + require.NoError(t, agentErr) + process, processErr := NewStressAgentProcess(agent, 200+index) + require.NoError(t, processErr) + values = append(values, StressRSSSeries{Process: process, EndRSSBytes: [3]uint64{100, 100, 100}}) + } + err = ValidateStressSoakTrendForProfile(profile, StressSoakTrendEvidence{Series: mutate(values)}) + require.Error(t, err) + } +} + +func mustProfile(t *testing.T, name contract.ProfileName) contract.Profile { + t.Helper() + profile, err := contract.ProfileByName(string(name)) + require.NoError(t, err) + return profile +} diff --git a/integration/agentcompat/internal/scenario/stress_sqlite_drain.go b/integration/agentcompat/internal/scenario/stress_sqlite_drain.go new file mode 100644 index 00000000..8885a962 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_sqlite_drain.go @@ -0,0 +1,87 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "context" + "errors" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +var ErrStressSQLiteJournalNotDrained = errors.New("stress dashboard sqlite journal is not drained") + +type stressSQLiteHoldControl interface { + ArmSQLiteHold(context.Context) (client.SQLiteHoldReceipt, error) + WaitForSQLiteHold(context.Context, client.SQLiteHoldReceipt, client.SQLiteHoldState) (client.SQLiteHoldReceipt, error) + ReleaseSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error) + AbortSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error) +} + +type stressSQLiteJournalWatch interface { + Wait(context.Context) error + Close() error +} + +type stressSQLiteJournalWatchOpener func(string) (stressSQLiteJournalWatch, error) + +func drainStressSQLiteJournal(ctx context.Context, control stressSQLiteHoldControl, writer func(context.Context) error, journalPath string, openWatch stressSQLiteJournalWatchOpener) error { + receipt, err := control.ArmSQLiteHold(ctx) + if err != nil { + return err + } + writerContext, cancelWriter := context.WithCancel(ctx) + writerDone := make(chan error, 1) + go func() { writerDone <- writer(writerContext) }() + abort := func(cause error) error { + _, abortErr := control.AbortSQLiteHold(context.WithoutCancel(ctx), receipt) + cancelWriter() + return errors.Join(cause, abortErr, <-writerDone) + } + selected, err := control.WaitForSQLiteHold(ctx, receipt, client.SQLiteHoldStateSelected) + if err != nil { + return abort(err) + } + finalizing, err := control.WaitForSQLiteHold(ctx, selected, client.SQLiteHoldStateFinalizing) + if err != nil { + return abort(err) + } + watch, err := openWatch(journalPath) + if err != nil { + return abort(err) + } + if _, err := control.ReleaseSQLiteHold(ctx, finalizing); err != nil { + return errors.Join(abort(err), watch.Close()) + } + waitErr := watch.Wait(ctx) + if waitErr != nil { + cancelWriter() + } + writerErr := <-writerDone + cancelWriter() + return errors.Join(waitErr, writerErr, watch.Close()) +} + +func drainStressDashboardSQLiteJournal(ctx context.Context, fixture *heldSessionSetRealFixture) error { + journalPath := fixture.dashboard.DatabasePath() + "-journal" + return drainStressSQLiteJournal(ctx, fixture.controlPAT.Client, func(writerContext context.Context) error { + _, err := fixture.controlPAT.Client.IOStreamState(writerContext) + return err + }, journalPath, func(path string) (stressSQLiteJournalWatch, error) { + return processharness.OpenSQLiteJournalWatch(path) + }) +} + +func observeStressDashboardSQLiteJournal(path string) func(context.Context, processharness.Sample) error { + return func(_ context.Context, sample processharness.Sample) error { + held, err := processharness.ProcessHasOpenPath(sample.PID, path) + if err != nil { + return err + } + if held { + return ErrStressSQLiteJournalNotDrained + } + return nil + } +} diff --git a/integration/agentcompat/internal/scenario/stress_sqlite_drain_test.go b/integration/agentcompat/internal/scenario/stress_sqlite_drain_test.go new file mode 100644 index 00000000..666afacb --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_sqlite_drain_test.go @@ -0,0 +1,174 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +type stressSQLiteHoldControlProbe struct { + mu sync.Mutex + calls []string + started chan struct{} + done chan struct{} + once sync.Once +} + +func newStressSQLiteHoldControlProbe() *stressSQLiteHoldControlProbe { + return &stressSQLiteHoldControlProbe{started: make(chan struct{}), done: make(chan struct{})} +} + +func (probe *stressSQLiteHoldControlProbe) record(call string) { + probe.mu.Lock() + defer probe.mu.Unlock() + probe.calls = append(probe.calls, call) +} + +func (probe *stressSQLiteHoldControlProbe) ArmSQLiteHold(context.Context) (client.SQLiteHoldReceipt, error) { + probe.record("arm") + return client.SQLiteHoldReceipt{ID: "ERERERERERERERERERERERERERERERERERERERERERE", State: client.SQLiteHoldStateArmed}, nil +} + +func (probe *stressSQLiteHoldControlProbe) WaitForSQLiteHold(ctx context.Context, receipt client.SQLiteHoldReceipt, target client.SQLiteHoldState) (client.SQLiteHoldReceipt, error) { + if target == client.SQLiteHoldStateSelected { + select { + case <-probe.started: + case <-ctx.Done(): + return client.SQLiteHoldReceipt{}, ctx.Err() + } + } + probe.record("wait-" + string(target)) + receipt.State = target + return receipt, nil +} + +func (probe *stressSQLiteHoldControlProbe) ReleaseSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error) { + probe.record("release") + probe.once.Do(func() { close(probe.done) }) + return client.SQLiteHoldReceipt{State: client.SQLiteHoldStateReleased}, nil +} + +func (probe *stressSQLiteHoldControlProbe) AbortSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error) { + probe.record("abort") + probe.once.Do(func() { close(probe.done) }) + return client.SQLiteHoldReceipt{State: client.SQLiteHoldStateAborted}, nil +} + +type stressSQLiteJournalWatchProbe struct { + control *stressSQLiteHoldControlProbe +} + +func (watch stressSQLiteJournalWatchProbe) Wait(ctx context.Context) error { + select { + case <-watch.control.done: + watch.control.record("watch-wait") + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (watch stressSQLiteJournalWatchProbe) Close() error { + watch.control.record("watch-close") + return nil +} + +func TestDrainStressSQLiteJournalOrdersHoldLifecycleBeforeCompletion(t *testing.T) { + // Given + control := newStressSQLiteHoldControlProbe() + journalPath := filepath.Join(t.TempDir(), "dashboard.sqlite-journal") + + // When + err := drainStressSQLiteJournal(t.Context(), control, func(ctx context.Context) error { + control.record("writer-start") + close(control.started) + select { + case <-control.done: + control.record("writer-complete") + return nil + case <-ctx.Done(): + return ctx.Err() + } + }, journalPath, func(path string) (stressSQLiteJournalWatch, error) { + require.Equal(t, journalPath, path) + control.record("watch-open") + return stressSQLiteJournalWatchProbe{control: control}, nil + }) + + // Then + require.NoError(t, err) + control.mu.Lock() + calls := append([]string(nil), control.calls...) + control.mu.Unlock() + require.Less(t, callIndex(calls, "arm"), callIndex(calls, "writer-start")) + require.Less(t, callIndex(calls, "writer-start"), callIndex(calls, "wait-selected")) + require.Less(t, callIndex(calls, "wait-selected"), callIndex(calls, "wait-finalizing")) + require.Less(t, callIndex(calls, "wait-finalizing"), callIndex(calls, "watch-open")) + require.Less(t, callIndex(calls, "watch-open"), callIndex(calls, "release")) + require.Less(t, callIndex(calls, "release"), callIndex(calls, "watch-wait")) + require.Less(t, callIndex(calls, "release"), callIndex(calls, "writer-complete")) + require.Less(t, callIndex(calls, "watch-wait"), callIndex(calls, "watch-close")) +} + +func TestDrainStressSQLiteJournalAbortsWriterWhenWatchCannotOpen(t *testing.T) { + // Given + control := newStressSQLiteHoldControlProbe() + watchErr := errors.New("watch unavailable") + + // When + err := drainStressSQLiteJournal(t.Context(), control, func(ctx context.Context) error { + close(control.started) + select { + case <-control.done: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }, filepath.Join(t.TempDir(), "dashboard.sqlite-journal"), func(string) (stressSQLiteJournalWatch, error) { + return nil, watchErr + }) + + // Then + require.ErrorIs(t, err, watchErr) + control.mu.Lock() + calls := append([]string(nil), control.calls...) + control.mu.Unlock() + require.Contains(t, calls, "abort") +} + +func TestObserveStressDashboardSQLiteJournalRejectsFirstHeldSample(t *testing.T) { + // Given + path := filepath.Join(t.TempDir(), "dashboard.sqlite-journal") + require.NoError(t, os.WriteFile(path, []byte("journal"), 0o600)) + file, err := os.Open(path) + require.NoError(t, err) + observer := observeStressDashboardSQLiteJournal(path) + + // When + heldErr := observer(t.Context(), processharness.Sample{PID: os.Getpid()}) + require.NoError(t, file.Close()) + releasedErr := observer(t.Context(), processharness.Sample{PID: os.Getpid()}) + + // Then + require.ErrorIs(t, heldErr, ErrStressSQLiteJournalNotDrained) + require.NoError(t, releasedErr) +} + +func callIndex(calls []string, wanted string) int { + for index, call := range calls { + if call == wanted { + return index + } + } + return len(calls) +}