diff --git a/.ai/TASK_REFERENCE.md b/.ai/TASK_REFERENCE.md index 48d1c634..fb419398 100644 --- a/.ai/TASK_REFERENCE.md +++ b/.ai/TASK_REFERENCE.md @@ -1130,6 +1130,53 @@ Executes a command with arguments. --- +### run_network_disruption + +Applies or heals network disruptions (partitions, isolations, shaping) on a Kurtosis devnet via the [disruptoor](https://github.com/ethpandaops/disruptoor) HTTP API. Waits for the API to report healthy, performs the action, then reads back the applied state. Partition/isolation/shaping entries are passed to disruptoor verbatim (wire format of `PUT /v1/state`); disruptoor validation errors are surfaced in the task failure. + +**Config:** +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `disruptoorUrl` | string | required | Base URL of the disruptoor HTTP API (e.g. `http://disruptoor:7700`) | +| `action` | string | "set" | `set` (replace whole state), `update` (merge entries by name), `clear` (heal everything) | +| `partitions` | array[object] | [] | Partition entries: `name`, `groups` (2+ disjoint selectors), optional `scope`, `symmetric` | +| `isolations` | array[object] | [] | Isolation entries: `name`, `target` selector cut off from the rest of the enclave (multi-container targets are isolated as a group), optional `scope` | +| `shaping` | array[object] | [] | Shaping entries: `name`, `target`, `delay`/`jitter`/`loss`/`bandwidth`, `scope: [include_control]` | +| `removeNames` | array[string] | [] | Entry names to remove before merging (`update` action only) | +| `awaitApiTimeout` | duration | 30s | How long to wait for the API to report healthy before acting (0 = act immediately) | +| `pollInterval` | duration | 2s | Interval between health probes | +| `requestTimeout` | duration | 10s | Timeout for a single HTTP request | + +Selectors are ethereum-package label matches (e.g. `{node-index: 1, client-type: beacon}`); `scope` values are `cl_p2p`, `el_p2p`, `include_control` (default `[cl_p2p, el_p2p]`; `include_control` also cuts RPC/engine/metrics/VC-CL traffic). + +**Outputs:** +| Variable | Type | Description | +|----------|------|-------------| +| `appliedState` | object | Disruptoor state after the action (reflects applied reality) | +| `partitionCount` | int | Active partitions after the action | +| `isolationCount` | int | Active isolations after the action | +| `shapingCount` | int | Active shaping rules after the action | + +**Example:** +```yaml +- name: run_network_disruption + title: "Black out node 1's beacon client" + config: + disruptoorUrl: "http://disruptoor:7700" + isolations: + - name: blackout-target-cl + target: { node-index: 1, client-type: beacon } + scope: [cl_p2p, el_p2p, include_control] + +# heal (also use as a cleanupTask): +- name: run_network_disruption + config: + disruptoorUrl: "http://disruptoor:7700" + action: clear +``` + +--- + ### run_spamoor_scenario Runs a spamoor stress testing scenario. diff --git a/go.mod b/go.mod index 9c626468..ae521193 100644 --- a/go.mod +++ b/go.mod @@ -34,6 +34,7 @@ require ( github.com/sirupsen/logrus v1.9.4 github.com/spf13/cobra v1.10.2 github.com/spf13/pflag v1.0.10 + github.com/stretchr/testify v1.11.1 github.com/swaggo/http-swagger v1.3.4 github.com/swaggo/swag v1.16.6 github.com/tyler-smith/go-bip39 v1.1.0 @@ -60,6 +61,7 @@ require ( github.com/consensys/gnark-crypto v0.20.1 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect github.com/crate-crypto/go-eth-kzg v1.5.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/deckarep/golang-set/v2 v2.6.0 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect github.com/emicklei/dot v1.6.4 // indirect @@ -94,6 +96,7 @@ require ( github.com/pk910/dynamic-ssz v1.3.2 // indirect github.com/pk910/hashtree-bindings v0.2.2 // indirect github.com/pkg/errors v0.9.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/procfs v0.20.1 // indirect github.com/protolambda/bls12-381-util v0.1.0 // indirect github.com/r3labs/sse/v2 v2.10.0 // indirect diff --git a/pkg/tasks/run_network_disruption/README.md b/pkg/tasks/run_network_disruption/README.md new file mode 100644 index 00000000..16e083ff --- /dev/null +++ b/pkg/tasks/run_network_disruption/README.md @@ -0,0 +1,166 @@ +# `run_network_disruption` Task + +## Description + +The `run_network_disruption` task drives a [disruptoor](https://github.com/ethpandaops/disruptoor) instance to apply or heal network disruptions — hard partitions, single-target isolations, and traffic shaping — on a Kurtosis-launched devnet. + +### Task Behavior + +- Waits for the disruptoor API to report healthy (configurable timeout), then performs the configured action. +- Partition, isolation, and shaping entries are passed through to disruptoor **verbatim** (the wire format of `PUT /v1/state`, see the [disruptoor JSON schema](https://github.com/ethpandaops/disruptoor/blob/master/schemas/v1-state.json)). Assertoor only checks that every entry carries a unique `name`; everything else is validated by disruptoor, whose error messages are surfaced in the task failure. +- Task outputs are read back via `GET /v1/state` after the action, which reflects the *applied* state. +- The task completes as soon as the disruption is applied (disruptoor applies synchronously); pair it with `check_*` tasks to assert the network effects, and put a `clear` invocation in `cleanupTasks` so an aborted test heals the network. + +### Actions + +- **`set`** (default): Replace the entire disruptoor state with the configured entries. Anything previously active that is not part of this request is healed. +- **`update`**: Read-merge-write. Entries named in `removeNames` are dropped from the current state, then each configured entry replaces its same-name predecessor or is appended. The write is guarded with `If-Match` and retried when a concurrent writer wins the race. Use this to compose disruptions across tasks (e.g. keep a baseline jitter while toggling a blackout). +- **`clear`**: Heal everything. + +## Configuration Parameters + +- **`disruptoorUrl`**:\ + Base URL of the disruptoor HTTP API (e.g. `http://disruptoor:7700`). Required. + +- **`action`**:\ + Action to perform: `set`, `update`, or `clear`. Default: `set`. + +- **`partitions`**:\ + Disruptoor partition entries. Each splits the enclave into 2+ disjoint groups; traffic crossing group boundaries is dropped. Fields: `name`, `groups` (list of selectors), optional `scope`, optional `symmetric`. + +- **`isolations`**:\ + Disruptoor isolation entries. Each cuts the containers matched by its `target` selector off from **the rest of the enclave** — the counterparty group is computed by disruptoor at apply time, so it never needs to be enumerated and stays correct when the topology changes. A target matching multiple containers is isolated *as a group* (traffic among its members keeps flowing); declare one isolation per container to black out several containers individually. Fields: `name`, `target` (selector), optional `scope`. + +- **`shaping`**:\ + Disruptoor shaping entries: per-target `delay`/`jitter`/`loss`/`bandwidth` degradation. Requires `scope: [include_control]` acknowledgement (disruptoor v0 shapes all egress traffic). + +- **`removeNames`**:\ + Entry names to remove from the current state before merging. Only valid with `action: update`. + +- **`awaitApiTimeout`**:\ + How long to wait for the disruptoor API to report healthy before acting. `0` acts immediately. Default: `30s`. + +- **`pollInterval`**:\ + Interval between health probes while waiting for the API. Default: `2s`. + +- **`requestTimeout`**:\ + Timeout for a single HTTP request. Default: `10s`. + +### Selectors and scopes + +Group and target selectors are label matches against the enclave's containers; keys without a dot get the `com.kurtosistech.custom.ethereum-package.` prefix. Common keys on ethereum-package devnets: `node-index` (1-based participant index) and `client-type` (`beacon`, `execution`, `validator`). Multiple values within a key OR together; multiple keys AND together. + +`scope` selects the port classes a disruption bites on: `cl_p2p`, `el_p2p` (the default pair), and `include_control` as an explicit opt-in to also cut RPC/engine/metrics/VC↔CL traffic. Without `include_control`, tests keep their visibility into the disrupted node. + +## Outputs + +- **`appliedState`**:\ + The disruptoor state after the action (object; reflects applied reality). + +- **`partitionCount`** / **`isolationCount`** / **`shapingCount`**:\ + Number of active entries of each kind after the action. + +## Examples + +### Fully black out one node's beacon client, then heal + +Cuts participant 1's CL off from everything — other participants *and* its own execution/validator client: + +```yaml +- name: run_network_disruption + title: "Black out the target beacon node" + config: + disruptoorUrl: "http://disruptoor:7700" + isolations: + - name: blackout-target-cl + target: { node-index: 1, client-type: beacon } + scope: [cl_p2p, el_p2p, include_control] + +# ... assert the network effects with check_* tasks ... + +- name: run_network_disruption + title: "Heal the blackout" + config: + disruptoorUrl: "http://disruptoor:7700" + action: clear +``` + +### Isolate a whole participant + +Without `client-type`, the target matches the participant's CL, EL, and VC together — they keep talking to *each other* but lose the rest of the network. Useful for "node offline" scenarios where the stack itself stays coherent: + +```yaml +- name: run_network_disruption + title: "Take participant 2 off the network" + config: + disruptoorUrl: "http://disruptoor:7700" + isolations: + - name: offline-node-2 + target: { node-index: 2 } +``` + +### Variable-driven targets + +Task `config` is static YAML; to build entries from test variables, set the whole field via a `configVars` jq expression: + +```yaml +- name: run_network_disruption + title: "Black out the configured participant" + configVars: + disruptoorUrl: "disruptoorApiUrl" + isolations: >- + | [{ + name: "assertoor-blackout-target-cl", + target: {"node-index": (.targetParticipantIndex | tonumber), "client-type": "beacon"}, + scope: ["cl_p2p", "el_p2p", "include_control"] + }] + config: {} +``` + +### Two-way network split + +```yaml +- name: run_network_disruption + title: "Split the network in half" + config: + disruptoorUrl: "http://disruptoor:7700" + partitions: + - name: fork-split + groups: + - { node-index: [1, 2] } + - { node-index: [3, 4] } + scope: [cl_p2p, el_p2p] +``` + +### Compose disruptions with update + +Keep a baseline jitter active while toggling a blackout on and off: + +```yaml +- name: run_network_disruption + title: "Add blackout on top of existing disruptions" + config: + disruptoorUrl: "http://disruptoor:7700" + action: update + isolations: + - name: blackout-node-3 + target: { node-index: 3 } + +- name: run_network_disruption + title: "Remove only the blackout" + config: + disruptoorUrl: "http://disruptoor:7700" + action: update + removeNames: [blackout-node-3] +``` + +### Cleanup task + +```yaml +cleanupTasks: + - name: run_network_disruption + title: "Heal all network disruptions" + config: + disruptoorUrl: "http://disruptoor:7700" + action: clear +``` diff --git a/pkg/tasks/run_network_disruption/config.go b/pkg/tasks/run_network_disruption/config.go new file mode 100644 index 00000000..6e23cafc --- /dev/null +++ b/pkg/tasks/run_network_disruption/config.go @@ -0,0 +1,140 @@ +package runnetworkdisruption + +import ( + "fmt" + "net/url" + "strings" + "time" + + "github.com/ethpandaops/assertoor/pkg/helper" +) + +// Action selects what the task does against the disruptoor API. +type Action string + +const ( + // ActionSet replaces the entire disruptoor state with the configured entries. + ActionSet Action = "set" + // ActionUpdate merges the configured entries into the current state by name. + ActionUpdate Action = "update" + // ActionClear heals all active disruptions. + ActionClear Action = "clear" +) + +// Config holds the task configuration for driving a disruptoor instance: +// applying network partitions, isolations, and shaping rules to a +// Kurtosis-launched devnet, or healing them again. +// +// Partition, isolation, and shaping entries are passed through to disruptoor +// verbatim (wire format of PUT /v1/state, see the disruptoor JSON schema). +// Assertoor only checks that every entry carries a name — everything else is +// validated server-side so new disruptoor fields work without assertoor +// changes. +type Config struct { + DisruptoorURL string `yaml:"disruptoorUrl" json:"disruptoorUrl" require:"A" desc:"Base URL of the disruptoor HTTP API (e.g. http://disruptoor:7700)."` + Action Action `yaml:"action" json:"action" desc:"Action to perform: set (replace the whole disruptoor state), update (merge entries by name into the current state), clear (heal everything)."` + Partitions []map[string]any `yaml:"partitions" json:"partitions,omitempty" desc:"Disruptoor partition entries: each splits the enclave into 2+ disjoint groups."` + Isolations []map[string]any `yaml:"isolations" json:"isolations,omitempty" desc:"Disruptoor isolation entries: each cuts the containers matched by its target selector off from the rest of the enclave (the counterparty group is computed by disruptoor; a multi-container target is isolated as a group)."` + Shaping []map[string]any `yaml:"shaping" json:"shaping,omitempty" desc:"Disruptoor shaping entries: per-target delay/jitter/loss/bandwidth degradation."` + RemoveNames []string `yaml:"removeNames" json:"removeNames,omitempty" desc:"Entry names to remove from the current state before merging (update action only)."` + AwaitAPITimeout helper.Duration `yaml:"awaitApiTimeout" json:"awaitApiTimeout" desc:"How long to wait for the disruptoor API to report healthy before acting (0 = act immediately)."` + PollInterval helper.Duration `yaml:"pollInterval" json:"pollInterval" desc:"Interval between health probes while waiting for the API."` + RequestTimeout helper.Duration `yaml:"requestTimeout" json:"requestTimeout" desc:"Timeout for a single HTTP request."` +} + +// DefaultConfig returns a Config with default values. +func DefaultConfig() Config { + return Config{ + Action: ActionSet, + AwaitAPITimeout: helper.Duration{Duration: 30 * time.Second}, + PollInterval: helper.Duration{Duration: 2 * time.Second}, + RequestTimeout: helper.Duration{Duration: 10 * time.Second}, + } +} + +// Validate validates the configuration. +func (c *Config) Validate() error { + if c.DisruptoorURL == "" { + return fmt.Errorf("disruptoorUrl is required") + } + + parsed, err := url.Parse(c.DisruptoorURL) + if err != nil { + return fmt.Errorf("invalid disruptoorUrl %q: %w", c.DisruptoorURL, err) + } + + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return fmt.Errorf("invalid disruptoorUrl %q: scheme must be http or https", c.DisruptoorURL) + } + + c.DisruptoorURL = strings.TrimRight(c.DisruptoorURL, "/") + + c.Action = Action(strings.ToLower(string(c.Action))) + + entryCount := len(c.Partitions) + len(c.Isolations) + len(c.Shaping) + + switch c.Action { + case ActionSet: + if entryCount == 0 { + return fmt.Errorf("set with no partitions/isolations/shaping would clear everything; use action: clear instead") + } + case ActionUpdate: + if entryCount == 0 && len(c.RemoveNames) == 0 { + return fmt.Errorf("update requires at least one partition/isolation/shaping entry or removeNames") + } + case ActionClear: + if entryCount > 0 { + return fmt.Errorf("clear does not take partition/isolation/shaping entries") + } + default: + return fmt.Errorf("invalid action %q, must be one of: set, update, clear", c.Action) + } + + if len(c.RemoveNames) > 0 && c.Action != ActionUpdate { + return fmt.Errorf("removeNames is only valid with action: update") + } + + if err := validateEntryNames("partitions", c.Partitions); err != nil { + return err + } + + if err := validateEntryNames("isolations", c.Isolations); err != nil { + return err + } + + if err := validateEntryNames("shaping", c.Shaping); err != nil { + return err + } + + if c.AwaitAPITimeout.Duration > 0 && c.PollInterval.Duration <= 0 { + return fmt.Errorf("pollInterval must be positive") + } + + if c.RequestTimeout.Duration <= 0 { + return fmt.Errorf("requestTimeout must be positive") + } + + return nil +} + +// validateEntryNames checks that every entry in a passthrough list carries a +// unique, non-empty name. Names are what disruptoor keys entries on and what +// the update action merges by, so they must be present client-side. +func validateEntryNames(kind string, entries []map[string]any) error { + seen := make(map[string]bool, len(entries)) + + for i, entry := range entries { + name, ok := entry["name"].(string) + if !ok || name == "" { + return fmt.Errorf("%s[%d]: name is required", kind, i) + } + + if seen[name] { + return fmt.Errorf("%s[%d]: duplicate name %q", kind, i, name) + } + + seen[name] = true + } + + return nil +} diff --git a/pkg/tasks/run_network_disruption/config_test.go b/pkg/tasks/run_network_disruption/config_test.go new file mode 100644 index 00000000..3e016441 --- /dev/null +++ b/pkg/tasks/run_network_disruption/config_test.go @@ -0,0 +1,157 @@ +package runnetworkdisruption + +import ( + "testing" + + "github.com/ethpandaops/assertoor/pkg/vars" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestConfigValidate(t *testing.T) { + isolation := map[string]any{testKeyName: testNameBlackout, testKeyTarget: map[string]any{testKeyNodeIndex: 1}} + + tests := []struct { + name string + mutate func(c *Config) + wantErr string + }{ + { + name: "valid set with isolation", + mutate: func(c *Config) { + c.Isolations = []map[string]any{isolation} + }, + }, + { + name: "missing url", + mutate: func(c *Config) { c.DisruptoorURL = "" }, + wantErr: "disruptoorUrl is required", + }, + { + name: "non-http url", + mutate: func(c *Config) { + c.DisruptoorURL = "ftp://disruptoor:7700" + c.Isolations = []map[string]any{isolation} + }, + wantErr: "scheme must be http or https", + }, + { + name: "set without entries", + mutate: func(_ *Config) {}, + wantErr: "use action: clear instead", + }, + { + name: "clear with entries", + mutate: func(c *Config) { + c.Action = ActionClear + c.Isolations = []map[string]any{isolation} + }, + wantErr: "clear does not take", + }, + { + name: "clear without entries", + mutate: func(c *Config) { c.Action = ActionClear }, + }, + { + name: "update without entries or removeNames", + mutate: func(c *Config) { c.Action = ActionUpdate }, + wantErr: "update requires at least one", + }, + { + name: "update with removeNames only", + mutate: func(c *Config) { + c.Action = ActionUpdate + c.RemoveNames = []string{testNameBlackout} + }, + }, + { + name: "removeNames with set", + mutate: func(c *Config) { + c.Isolations = []map[string]any{isolation} + c.RemoveNames = []string{testNameBlackout} + }, + wantErr: "removeNames is only valid with action: update", + }, + { + name: "unknown action", + mutate: func(c *Config) { + c.Action = "heal" + }, + wantErr: "invalid action", + }, + { + name: "action is case-insensitive", + mutate: func(c *Config) { + c.Action = "CLEAR" + }, + }, + { + name: "entry without name", + mutate: func(c *Config) { + c.Isolations = []map[string]any{{testKeyTarget: map[string]any{testKeyNodeIndex: 1}}} + }, + wantErr: "isolations[0]: name is required", + }, + { + name: "duplicate entry names", + mutate: func(c *Config) { + c.Partitions = []map[string]any{ + {testKeyName: "x", testKeyGroups: []any{}}, + {testKeyName: "x", testKeyGroups: []any{}}, + } + }, + wantErr: `partitions[1]: duplicate name "x"`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config := DefaultConfig() + config.DisruptoorURL = "http://disruptoor:7700" + tt.mutate(&config) + + err := config.Validate() + if tt.wantErr == "" { + assert.NoError(t, err) + return + } + + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + }) + } +} + +// TestConfigFromConfigVars exercises the playbook usage pattern: the +// isolations list is built by a jq expression in configVars and lands in the +// Config via the vars system's YAML round-trip. +func TestConfigFromConfigVars(t *testing.T) { + taskVars := vars.NewVariables(nil) + taskVars.SetVar("disruptoorApiUrl", "http://disruptoor:7700") + taskVars.SetVar("targetParticipantIndex", 1) + + config := DefaultConfig() + require.NoError(t, taskVars.ConsumeVars(&config, map[string]string{ + "disruptoorUrl": "disruptoorApiUrl", + "isolations": `| [{ + name: "assertoor-blackout-target-cl", + target: {"node-index": (.targetParticipantIndex | tonumber), "client-type": "beacon"}, + scope: ["cl_p2p", "el_p2p", "include_control"] + }]`, + })) + require.NoError(t, config.Validate()) + + assert.Equal(t, "http://disruptoor:7700", config.DisruptoorURL) + require.Len(t, config.Isolations, 1) + assert.Equal(t, "assertoor-blackout-target-cl", config.Isolations[0][testKeyName]) + assert.Equal(t, map[string]any{testKeyNodeIndex: 1, "client-type": "beacon"}, config.Isolations[0][testKeyTarget]) +} + +func TestConfigValidateTrimsTrailingSlash(t *testing.T) { + config := DefaultConfig() + config.DisruptoorURL = "http://disruptoor:7700/" + config.Action = ActionClear + + require.NoError(t, config.Validate()) + assert.Equal(t, "http://disruptoor:7700", config.DisruptoorURL) +} diff --git a/pkg/tasks/run_network_disruption/task.go b/pkg/tasks/run_network_disruption/task.go new file mode 100644 index 00000000..a5277980 --- /dev/null +++ b/pkg/tasks/run_network_disruption/task.go @@ -0,0 +1,488 @@ +package runnetworkdisruption + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "github.com/ethpandaops/assertoor/pkg/types" + "github.com/ethpandaops/assertoor/pkg/vars" + "github.com/sirupsen/logrus" +) + +const ( + // maxResponseSize caps disruptoor API response bodies; state documents + // are tiny, so 10MB is a generous safety bound. + maxResponseSize = 10 * 1024 * 1024 + + // maxUpdateAttempts bounds the read-merge-write retry loop when a + // concurrent writer invalidates our ETag between GET and PUT. + maxUpdateAttempts = 3 + + stateKeyPartitions = "partitions" + stateKeyIsolations = "isolations" + stateKeyShaping = "shaping" + + outputTypeInt = "int" +) + +// Compile-time interface compliance check. +var _ types.Task = (*Task)(nil) + +var ( + TaskName = "run_network_disruption" + TaskDescriptor = &types.TaskDescriptor{ + Name: TaskName, + Description: "Applies or heals network disruptions (partitions, isolations, shaping) via the disruptoor API.", + Category: "utility", + Config: DefaultConfig(), + Outputs: []types.TaskOutputDefinition{ + { + Name: "appliedState", + Type: "object", + Description: "The disruptoor state applied after the action (reflects reality, fetched via GET /v1/state).", + }, + { + Name: "partitionCount", + Type: outputTypeInt, + Description: "Number of active partitions after the action.", + }, + { + Name: "isolationCount", + Type: outputTypeInt, + Description: "Number of active isolations after the action.", + }, + { + Name: "shapingCount", + Type: outputTypeInt, + Description: "Number of active shaping rules after the action.", + }, + }, + NewTask: NewTask, + } + + errStalePut = fmt.Errorf("state changed between read and write") +) + +// Task implements the run_network_disruption task. +type Task struct { + ctx *types.TaskContext + options *types.TaskOptions + config Config + logger logrus.FieldLogger + httpClient *http.Client +} + +// NewTask creates a new run_network_disruption task. +func NewTask(ctx *types.TaskContext, options *types.TaskOptions) (types.Task, error) { + return &Task{ + ctx: ctx, + options: options, + logger: ctx.Logger.GetLogger(), + }, nil +} + +// Config returns the task configuration. +func (t *Task) Config() interface{} { + return t.config +} + +// Timeout returns the task timeout. +func (t *Task) Timeout() time.Duration { + return t.options.Timeout.Duration +} + +// LoadConfig loads and validates the task configuration. +func (t *Task) LoadConfig() error { + config := DefaultConfig() + + // Parse static config + if t.options.Config != nil { + if err := t.options.Config.Unmarshal(&config); err != nil { + return fmt.Errorf("error parsing task config for %v: %w", TaskName, err) + } + } + + // Load dynamic vars + if err := t.ctx.Vars.ConsumeVars(&config, t.options.ConfigVars); err != nil { + return err + } + + // Validate config + if err := config.Validate(); err != nil { + return err + } + + t.config = config + + t.httpClient = &http.Client{ + Timeout: config.RequestTimeout.Duration, + } + + return nil +} + +// Execute runs the task. +func (t *Task) Execute(ctx context.Context) error { + if t.config.AwaitAPITimeout.Duration > 0 { + t.ctx.ReportProgress(0, "Waiting for disruptoor API...") + + if err := t.awaitHealthy(ctx); err != nil { + return err + } + } + + t.ctx.ReportProgress(25, fmt.Sprintf("Performing %v action...", t.config.Action)) + + var err error + + switch t.config.Action { + case ActionClear: + err = t.clearState(ctx) + case ActionSet: + err = t.putState(ctx, t.buildDesiredState(), "") + case ActionUpdate: + err = t.updateState(ctx) + } + + if err != nil { + return err + } + + // GET reflects applied reality per the disruptoor API contract, so use + // it (rather than the PUT echo) as the source for the task outputs. + applied, _, err := t.getState(ctx) + if err != nil { + return fmt.Errorf("fetching applied state: %w", err) + } + + t.setOutputs(applied) + t.ctx.ReportProgress(100, fmt.Sprintf("Disruption %v action completed", t.config.Action)) + t.logger.Infof("disruptoor %v action applied (%d partitions, %d isolations, %d shaping rules active)", + t.config.Action, entryCount(applied, stateKeyPartitions), + entryCount(applied, stateKeyIsolations), entryCount(applied, stateKeyShaping)) + + return nil +} + +// awaitHealthy polls the disruptoor healthz endpoint until it responds OK or +// the configured timeout elapses. +func (t *Task) awaitHealthy(ctx context.Context) error { + deadline := time.Now().Add(t.config.AwaitAPITimeout.Duration) + + for { + err := t.probeHealth(ctx) + if err == nil { + return nil + } + + if time.Now().After(deadline) { + return fmt.Errorf("disruptoor API not healthy after %v: %w", t.config.AwaitAPITimeout.Duration, err) + } + + t.logger.Debugf("disruptoor API not healthy yet: %v", err) + + select { + case <-time.After(t.config.PollInterval.Duration): + case <-ctx.Done(): + return ctx.Err() + } + } +} + +func (t *Task) probeHealth(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, t.config.DisruptoorURL+"/v1/healthz", http.NoBody) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + resp, err := t.httpClient.Do(req) + if err != nil { + return err + } + + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("healthz returned status %d", resp.StatusCode) + } + + return nil +} + +// buildDesiredState assembles the wire-format state document from the +// configured passthrough entries. +func (t *Task) buildDesiredState() map[string]any { + state := make(map[string]any, 3) + if len(t.config.Partitions) > 0 { + state[stateKeyPartitions] = t.config.Partitions + } + + if len(t.config.Isolations) > 0 { + state[stateKeyIsolations] = t.config.Isolations + } + + if len(t.config.Shaping) > 0 { + state[stateKeyShaping] = t.config.Shaping + } + + return state +} + +// updateState merges the configured entries into the current disruptoor +// state: entries named in removeNames are dropped, then each configured +// entry replaces its same-name predecessor or is appended. The write is +// guarded with If-Match and retried when a concurrent writer wins the race. +func (t *Task) updateState(ctx context.Context) error { + for attempt := 1; ; attempt++ { + current, etag, err := t.getState(ctx) + if err != nil { + return fmt.Errorf("fetching current state: %w", err) + } + + merged := t.mergeState(current) + + err = t.putState(ctx, merged, etag) + if err == nil { + return nil + } + + if err != errStalePut || attempt >= maxUpdateAttempts { + return err + } + + t.logger.Warnf("disruptoor state changed concurrently, retrying update (attempt %d/%d)", attempt, maxUpdateAttempts) + } +} + +func (t *Task) mergeState(current map[string]any) map[string]any { + removeSet := make(map[string]bool, len(t.config.RemoveNames)) + for _, name := range t.config.RemoveNames { + removeSet[name] = true + } + + merged := make(map[string]any, 3) + + for _, key := range []string{stateKeyPartitions, stateKeyIsolations, stateKeyShaping} { + entries := stateEntries(current, key) + kept := make([]map[string]any, 0, len(entries)) + + for _, entry := range entries { + if name, ok := entry["name"].(string); ok && removeSet[name] { + continue + } + + kept = append(kept, entry) + } + + if len(kept) > 0 { + merged[key] = kept + } + } + + mergeEntries(merged, stateKeyPartitions, t.config.Partitions) + mergeEntries(merged, stateKeyIsolations, t.config.Isolations) + mergeEntries(merged, stateKeyShaping, t.config.Shaping) + + return merged +} + +func (t *Task) getState(ctx context.Context) (state map[string]any, etag string, err error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, t.config.DisruptoorURL+"/v1/state", http.NoBody) + if err != nil { + return nil, "", fmt.Errorf("failed to create request: %w", err) + } + + resp, err := t.httpClient.Do(req) + if err != nil { + return nil, "", fmt.Errorf("request failed: %w", err) + } + + defer resp.Body.Close() + + body, err := readBody(resp) + if err != nil { + return nil, "", err + } + + if resp.StatusCode != http.StatusOK { + return nil, "", apiError("GET /v1/state", resp.StatusCode, body) + } + + state = make(map[string]any, 3) + if err := json.Unmarshal(body, &state); err != nil { + return nil, "", fmt.Errorf("failed to parse state response: %w", err) + } + + return state, resp.Header.Get("ETag"), nil +} + +// putState PUTs the desired state, optionally guarded by an If-Match ETag. +// Returns errStalePut on a 412 so callers can re-read and retry. +func (t *Task) putState(ctx context.Context, state map[string]any, etag string) error { + encoded, err := json.Marshal(state) + if err != nil { + return fmt.Errorf("failed to encode state: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPut, t.config.DisruptoorURL+"/v1/state", bytes.NewReader(encoded)) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + + if etag != "" { + req.Header.Set("If-Match", etag) + } + + resp, err := t.httpClient.Do(req) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + + defer resp.Body.Close() + + body, err := readBody(resp) + if err != nil { + return err + } + + if resp.StatusCode == http.StatusPreconditionFailed { + return errStalePut + } + + if resp.StatusCode != http.StatusOK { + return apiError("PUT /v1/state", resp.StatusCode, body) + } + + return nil +} + +func (t *Task) clearState(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, t.config.DisruptoorURL+"/v1/state/clear", http.NoBody) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + resp, err := t.httpClient.Do(req) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + + defer resp.Body.Close() + + body, err := readBody(resp) + if err != nil { + return err + } + + if resp.StatusCode != http.StatusOK { + return apiError("POST /v1/state/clear", resp.StatusCode, body) + } + + return nil +} + +func (t *Task) setOutputs(applied map[string]any) { + if data, err := vars.GeneralizeData(applied); err != nil { + t.logger.Warnf("failed to generalize appliedState: %v", err) + } else { + t.ctx.Outputs.SetVar("appliedState", data) + } + + t.ctx.Outputs.SetVar("partitionCount", entryCount(applied, stateKeyPartitions)) + t.ctx.Outputs.SetVar("isolationCount", entryCount(applied, stateKeyIsolations)) + t.ctx.Outputs.SetVar("shapingCount", entryCount(applied, stateKeyShaping)) +} + +// mergeEntries replaces same-name entries in dst[key] with their configured +// counterparts and appends entries that aren't present yet. +func mergeEntries(dst map[string]any, key string, entries []map[string]any) { + if len(entries) == 0 { + return + } + + existing, _ := dst[key].([]map[string]any) + + for _, entry := range entries { + name, _ := entry["name"].(string) + replaced := false + + for i, cur := range existing { + if curName, ok := cur["name"].(string); ok && curName == name { + existing[i] = entry + replaced = true + + break + } + } + + if !replaced { + existing = append(existing, entry) + } + } + + dst[key] = existing +} + +// stateEntries extracts a list of entry objects from a decoded state +// document, tolerating both absent keys and unexpected shapes. +func stateEntries(state map[string]any, key string) []map[string]any { + raw, ok := state[key].([]any) + if !ok { + return nil + } + + out := make([]map[string]any, 0, len(raw)) + + for _, item := range raw { + if entry, ok := item.(map[string]any); ok { + out = append(out, entry) + } + } + + return out +} + +func entryCount(state map[string]any, key string) int { + switch entries := state[key].(type) { + case []any: + return len(entries) + case []map[string]any: + return len(entries) + default: + return 0 + } +} + +func readBody(resp *http.Response) ([]byte, error) { + body, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseSize+1)) + if err != nil { + return nil, fmt.Errorf("failed to read response body: %w", err) + } + + if len(body) > maxResponseSize { + return nil, fmt.Errorf("response body exceeds max size (%d bytes)", maxResponseSize) + } + + return body, nil +} + +// apiError surfaces the server-side error message when disruptoor rejects a +// request (its validation errors are precise and actionable). +func apiError(op string, status int, body []byte) error { + var errBody struct { + Error string `json:"error"` + } + + if err := json.Unmarshal(body, &errBody); err == nil && errBody.Error != "" { + return fmt.Errorf("%s returned status %d: %s", op, status, errBody.Error) + } + + return fmt.Errorf("%s returned status %d", op, status) +} diff --git a/pkg/tasks/run_network_disruption/task_test.go b/pkg/tasks/run_network_disruption/task_test.go new file mode 100644 index 00000000..2707a3b9 --- /dev/null +++ b/pkg/tasks/run_network_disruption/task_test.go @@ -0,0 +1,237 @@ +package runnetworkdisruption + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/helper" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Shared fixture keys/names so goconst doesn't trip over repeated literals. +const ( + testKeyName = "name" + testKeyTarget = "target" + testNameBlackout = "blackout" + testKeyNodeIndex = "node-index" + testKeyGroups = "groups" +) + +// fakeDisruptoor is a minimal in-memory stand-in for the disruptoor v1 API: +// it stores whatever state is PUT, serves it back on GET with a naive ETag, +// and honours If-Match / clear semantics. +type fakeDisruptoor struct { + state map[string]any + etag int + putCount int + rejects string // when non-empty, PUT fails 400 with this error message +} + +func (f *fakeDisruptoor) handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /v1/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("GET /v1/state", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("ETag", f.currentETag()) + _ = json.NewEncoder(w).Encode(f.state) + }) + mux.HandleFunc("PUT /v1/state", func(w http.ResponseWriter, r *http.Request) { + f.putCount++ + + if f.rejects != "" { + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]string{"error": f.rejects}) + + return + } + + if ifMatch := r.Header.Get("If-Match"); ifMatch != "" && ifMatch != f.currentETag() { + w.WriteHeader(http.StatusPreconditionFailed) + _ = json.NewEncoder(w).Encode(map[string]string{"error": "state changed"}) + + return + } + + desired := make(map[string]any, 3) + if err := json.NewDecoder(r.Body).Decode(&desired); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + + f.state = desired + f.etag++ + + _ = json.NewEncoder(w).Encode(f.state) + }) + mux.HandleFunc("POST /v1/state/clear", func(w http.ResponseWriter, _ *http.Request) { + f.state = map[string]any{} + f.etag++ + + _ = json.NewEncoder(w).Encode(map[string]string{"status": "cleared"}) + }) + + return mux +} + +func (f *fakeDisruptoor) currentETag() string { + return `"` + strconv.Itoa(f.etag) + `"` +} + +func newTestTask(t *testing.T, serverURL string, mutate func(c *Config)) *Task { + t.Helper() + + config := DefaultConfig() + config.DisruptoorURL = serverURL + config.AwaitAPITimeout = helper.Duration{Duration: 0} + mutate(&config) + require.NoError(t, config.Validate()) + + return &Task{ + config: config, + logger: logrus.New(), + httpClient: &http.Client{Timeout: 5 * time.Second}, + } +} + +func TestPutStateSendsConfiguredEntries(t *testing.T) { + fake := &fakeDisruptoor{state: map[string]any{}} + srv := httptest.NewServer(fake.handler()) + + defer srv.Close() + + task := newTestTask(t, srv.URL, func(c *Config) { + c.Isolations = []map[string]any{{ + testKeyName: testNameBlackout, + testKeyTarget: map[string]any{testKeyNodeIndex: 1, "client-type": "beacon"}, + "scope": []any{"cl_p2p", "el_p2p", "include_control"}, + }} + }) + + require.NoError(t, task.putState(context.Background(), task.buildDesiredState(), "")) + + entries := stateEntries(fake.state, stateKeyIsolations) + require.Len(t, entries, 1) + assert.Equal(t, testNameBlackout, entries[0][testKeyName]) +} + +func TestPutStateSurfacesServerError(t *testing.T) { + fake := &fakeDisruptoor{state: map[string]any{}, rejects: `target cannot be "all"`} + srv := httptest.NewServer(fake.handler()) + + defer srv.Close() + + task := newTestTask(t, srv.URL, func(c *Config) { + c.Isolations = []map[string]any{{testKeyName: "bad", testKeyTarget: "all"}} + }) + + err := task.putState(context.Background(), task.buildDesiredState(), "") + require.Error(t, err) + assert.Contains(t, err.Error(), `target cannot be "all"`) +} + +func TestUpdateMergesByNameAndRemoves(t *testing.T) { + fake := &fakeDisruptoor{state: map[string]any{ + stateKeyPartitions: []any{ + map[string]any{testKeyName: "keep-me", testKeyGroups: []any{"a", "b"}}, + }, + stateKeyIsolations: []any{ + map[string]any{testKeyName: "replace-me", testKeyTarget: map[string]any{testKeyNodeIndex: 1}}, + map[string]any{testKeyName: "remove-me", testKeyTarget: map[string]any{testKeyNodeIndex: 2}}, + }, + }} + srv := httptest.NewServer(fake.handler()) + + defer srv.Close() + + task := newTestTask(t, srv.URL, func(c *Config) { + c.Action = ActionUpdate + c.RemoveNames = []string{"remove-me"} + c.Isolations = []map[string]any{ + {testKeyName: "replace-me", testKeyTarget: map[string]any{testKeyNodeIndex: 3}}, + {testKeyName: "brand-new", testKeyTarget: map[string]any{testKeyNodeIndex: 4}}, + } + }) + + require.NoError(t, task.updateState(context.Background())) + + partitions := stateEntries(fake.state, stateKeyPartitions) + require.Len(t, partitions, 1) + assert.Equal(t, "keep-me", partitions[0][testKeyName]) + + isolations := stateEntries(fake.state, stateKeyIsolations) + require.Len(t, isolations, 2) + assert.Equal(t, "replace-me", isolations[0][testKeyName]) + assert.Equal(t, map[string]any{testKeyNodeIndex: float64(3)}, isolations[0][testKeyTarget]) + assert.Equal(t, "brand-new", isolations[1][testKeyName]) +} + +func TestClearStateHealsEverything(t *testing.T) { + fake := &fakeDisruptoor{state: map[string]any{ + stateKeyIsolations: []any{map[string]any{testKeyName: testNameBlackout}}, + }} + srv := httptest.NewServer(fake.handler()) + + defer srv.Close() + + task := newTestTask(t, srv.URL, func(c *Config) { + c.Action = ActionClear + }) + + require.NoError(t, task.clearState(context.Background())) + assert.Empty(t, fake.state) +} + +func TestUpdateRetriesOnStaleETag(t *testing.T) { + fake := &fakeDisruptoor{state: map[string]any{}} + // Bump the ETag underneath the first PUT to simulate a concurrent writer. + raced := false + mux := http.NewServeMux() + mux.Handle("/", fake.handler()) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut && !raced { + raced = true + f := fake + f.etag++ + } + + mux.ServeHTTP(w, r) + })) + + defer srv.Close() + + task := newTestTask(t, srv.URL, func(c *Config) { + c.Action = ActionUpdate + c.Isolations = []map[string]any{{testKeyName: testNameBlackout, testKeyTarget: map[string]any{testKeyNodeIndex: 1}}} + }) + + require.NoError(t, task.updateState(context.Background())) + require.Equal(t, 2, fake.putCount, "first PUT should 412, second should succeed") + + isolations := stateEntries(fake.state, stateKeyIsolations) + require.Len(t, isolations, 1) +} + +func TestProbeHealth(t *testing.T) { + fake := &fakeDisruptoor{state: map[string]any{}} + srv := httptest.NewServer(fake.handler()) + + defer srv.Close() + + task := newTestTask(t, srv.URL, func(c *Config) { + c.Action = ActionClear + }) + + require.NoError(t, task.probeHealth(context.Background())) + + srv.Close() + require.Error(t, task.probeHealth(context.Background())) +} diff --git a/pkg/tasks/tasks.go b/pkg/tasks/tasks.go index dfcb9352..1c1b1b44 100644 --- a/pkg/tasks/tasks.go +++ b/pkg/tasks/tasks.go @@ -46,6 +46,7 @@ import ( runcommand "github.com/ethpandaops/assertoor/pkg/tasks/run_command" runexternaltasks "github.com/ethpandaops/assertoor/pkg/tasks/run_external_tasks" runjavascript "github.com/ethpandaops/assertoor/pkg/tasks/run_javascript" + runnetworkdisruption "github.com/ethpandaops/assertoor/pkg/tasks/run_network_disruption" runshell "github.com/ethpandaops/assertoor/pkg/tasks/run_shell" runspamoorscenario "github.com/ethpandaops/assertoor/pkg/tasks/run_spamoor_scenario" runtaskbackground "github.com/ethpandaops/assertoor/pkg/tasks/run_task_background" @@ -102,6 +103,7 @@ var AvailableTaskDescriptors = []*types.TaskDescriptor{ runcommand.TaskDescriptor, runexternaltasks.TaskDescriptor, runjavascript.TaskDescriptor, + runnetworkdisruption.TaskDescriptor, runshell.TaskDescriptor, runspamoorscenario.TaskDescriptor, runtaskbackground.TaskDescriptor, diff --git a/playbooks/_index.yaml b/playbooks/_index.yaml index 6d74d84f..d05fd542 100644 --- a/playbooks/_index.yaml +++ b/playbooks/_index.yaml @@ -1,8 +1,8 @@ # Auto-generated playbook index -# Generated: 2026-07-07T07:50:04Z +# Generated: 2026-07-13T21:47:13Z # DO NOT EDIT MANUALLY - regenerate via `make generate-playbook-index`. -generated: 2026-07-07T07:50:04Z +generated: 2026-07-13T21:47:13Z folders: - path: api-compatibility name: API Compatibility @@ -219,6 +219,41 @@ playbooks: tags: - utility timeout: 4h + - file: dev/single-node-blackout-recovery.yaml + id: single-node-blackout-recovery + name: Single-Node Blackout & Recovery + description: |- + Fully isolates one participant's beacon node from the rest of a + Kurtosis-launched devnet, keeps the blackout active for a configurable + number of slots, heals it, then verifies the node re-syncs and the + network regains fresh finality. + + The blackout is a single disruptoor isolation applied through the + run_network_disruption task: the target beacon node loses ALL traffic — + p2p and control traffic to every other participant AND to its own + execution and validator client (scope `[cl_p2p, el_p2p, include_control]`). + Disruptoor computes the counterparty group as the complement of the + target, so no participant enumeration is needed and the playbook works + unchanged on any topology. This reproduces client behaviour under sudden + total network loss — e.g. a silent stall followed by a missed proposal + during catch-up. + + The blackout is gated on `minStartEpoch` so fork-activation devnets can + schedule it safely after the fork (set it to fork epoch + 2 or later). + The target participant is selected by its 1-based `node-index` label via + `targetParticipantIndex`. Choose a topology where the remaining nodes + retain a 2/3 finality majority (e.g. 4 nodes, blackout 1). A disruptoor + service is expected at `disruptoorApiUrl` (default + `http://disruptoor:7700`). + version: 1.1.0 + tags: + - disruptoor + - kurtosis + - blackout + - node-isolation + - recovery + - consensus + timeout: 120m - file: dev/synchronized-check.yaml id: synchronized-check name: Check client pairs are in sync @@ -655,6 +690,43 @@ playbooks: - queue - eip-8282 timeout: 1h + - file: gloas-dev/deploy-eip8282-contracts.yaml + id: deploy-eip8282-contracts + name: 'GLOAS: Deploy EIP-8282 builder request contracts (pre-fork)' + description: |- + Deploys the EIP-8282 builder deposit (request type 0x03) and builder exit + (request type 0x04) system contracts onto a running devnet BEFORE the GLOAS + fork activates. + + glamsterdam-devnet-7 deliberately omits these predeploys from genesis + (DEPLOY_EIP8282_CONTRACTS=false in the genesis generator) to exercise the + mainnet-like deployment workflow: from GLOAS activation onward every block + is INVALID unless both contracts exist, so they must be deployed as regular + transactions pre-fork. Intended to be registered as a startup test so the + contracts land right after the devnet comes up. + + Deployment goes through the deterministic CREATE2 factory (predeployed in + devnet genesis as a well-known contract) using the vanity-mined salts from + ethereum/execution-specs#3091 — CREATE2(factory, salt, initcode) yields the + canonical ...8282 addresses the clients expect. The factory REVERTS when + CREATE2 fails (including the already-deployed collision case), so a + successful receipt is proof of deployment. + + Idempotent: an eth_call precheck against the builder deposit contract's fee + getter detects existing code and skips deployment, so restarts of a + startup-scheduled run stay green. The exit contract cannot be prechecked the + same way — its constructor arms the slot-0 excess inhibitor, which makes its + fee getter revert by design until the first post-GLOAS system call — so its + deployment is asserted via the deploy receipt (both contracts only ever + deploy as a pair). + version: 1.0.0 + tags: + - gloas + - builder + - eip-8282 + - deployment + - prefork + timeout: 30m - file: gloas-dev/exit-builders.yaml id: exit-builders name: 'GLOAS: Exit builders deposited by prefork-queue-fill-public' diff --git a/playbooks/dev/single-node-blackout-recovery.yaml b/playbooks/dev/single-node-blackout-recovery.yaml new file mode 100644 index 00000000..2d9a7174 --- /dev/null +++ b/playbooks/dev/single-node-blackout-recovery.yaml @@ -0,0 +1,136 @@ +id: single-node-blackout-recovery +name: "Single-Node Blackout & Recovery" +description: | + Fully isolates one participant's beacon node from the rest of a + Kurtosis-launched devnet, keeps the blackout active for a configurable + number of slots, heals it, then verifies the node re-syncs and the + network regains fresh finality. + + The blackout is a single disruptoor isolation applied through the + run_network_disruption task: the target beacon node loses ALL traffic — + p2p and control traffic to every other participant AND to its own + execution and validator client (scope `[cl_p2p, el_p2p, include_control]`). + Disruptoor computes the counterparty group as the complement of the + target, so no participant enumeration is needed and the playbook works + unchanged on any topology. This reproduces client behaviour under sudden + total network loss — e.g. a silent stall followed by a missed proposal + during catch-up. + + The blackout is gated on `minStartEpoch` so fork-activation devnets can + schedule it safely after the fork (set it to fork epoch + 2 or later). + The target participant is selected by its 1-based `node-index` label via + `targetParticipantIndex`. Choose a topology where the remaining nodes + retain a 2/3 finality majority (e.g. 4 nodes, blackout 1). A disruptoor + service is expected at `disruptoorApiUrl` (default + `http://disruptoor:7700`). +version: 1.1.0 +tags: [disruptoor, kurtosis, blackout, node-isolation, recovery, consensus] +timeout: 120m +config: + disruptoorApiUrl: "http://disruptoor:7700" # Disruptoor HTTP API used to apply and clear the blackout. + targetParticipantIndex: 1 # 1-based participant index (node-index label) whose beacon node gets blacked out. + blackoutDurationSlots: 5 # Number of slots to keep the blackout active (5 slots = 30s on the 6s-slot minimal preset). + minStartEpoch: 3 # Do not start the blackout before this epoch (set to fork epoch + 2 or later on fork-activation devnets). + minimumClientCount: 2 # Minimum number of healthy clients required before starting the blackout. + minFinalizedEpochIncreaseAfterRecovery: 1 # Required finalized epoch increase after healing to prove fresh finality. + maxUnfinalizedEpochsAfterRecovery: 6 # Maximum finality lag allowed after healing. +tasks: +- name: run_network_disruption + title: "Wait for disruptoor and clear stale disruptions" + timeout: 5m + configVars: + disruptoorUrl: "disruptoorApiUrl" + config: + action: clear + awaitApiTimeout: 2m + +- name: check_clients_are_healthy + title: "Wait for all devnet clients to be healthy" + timeout: 20m + configVars: + minClientCount: "minimumClientCount" + config: + maxUnhealthyCount: 0 + +- name: check_consensus_finality + id: initial_finality + title: "Wait for initial finality" + timeout: 40m + config: + minFinalizedEpochs: 2 + maxUnfinalizedEpochs: 3 + +- name: check_consensus_slot_range + title: "Wait for the configured blackout start epoch" + timeout: 30m + configVars: + minEpochNumber: "minStartEpoch" + +- name: check_consensus_slot_range + id: blackout_start + title: "Capture blackout start slot" + timeout: 1m + config: {} + +- name: run_network_disruption + title: "Black out the target beacon node" + timeout: 3m + configVars: + disruptoorUrl: "disruptoorApiUrl" + # A single isolation: the target CL vs everything else (other + # participants AND its own EL + VC). include_control extends the cut + # beyond p2p to RPC/engine/API traffic. + isolations: >- + | [{ + name: "assertoor-blackout-target-cl", + target: {"node-index": (.targetParticipantIndex | tonumber), "client-type": "beacon"}, + scope: ["cl_p2p", "el_p2p", "include_control"] + }] + config: {} + +- name: check_consensus_slot_range + title: "Keep the blackout active for the configured duration" + timeout: 15m + configVars: + minSlotNumber: "| (.tasks.blackout_start.outputs.currentSlot | tonumber) + (.blackoutDurationSlots | tonumber)" + +- name: run_network_disruption + title: "Heal the blackout" + timeout: 2m + configVars: + disruptoorUrl: "disruptoorApiUrl" + config: + action: clear + +- name: check_clients_are_healthy + title: "Wait for all clients to report healthy after healing" + timeout: 10m + configVars: + minClientCount: "minimumClientCount" + config: + maxUnhealthyCount: 0 + +- name: check_consensus_sync_status + title: "Wait for all clients to report synced after healing" + timeout: 10m + config: + expectSyncing: false + expectMinPercent: 100 + +- name: check_consensus_finality + title: "Poll for fresh finality after recovery" + timeout: 25m + configVars: + # Require a fresh finalized checkpoint after the blackout. maxUnfinalizedEpochs + # alone can pass on stale pre-blackout finality when the current epoch is close. + minFinalizedEpochs: "| (.tasks.initial_finality.outputs.finalizedEpoch | tonumber) + (.minFinalizedEpochIncreaseAfterRecovery | tonumber)" + maxUnfinalizedEpochs: "maxUnfinalizedEpochsAfterRecovery" + +cleanupTasks: +- name: run_network_disruption + title: "Clear disruptoor state" + timeout: 2m + configVars: + disruptoorUrl: "disruptoorApiUrl" + config: + action: clear