diff --git a/.env.example b/.env.example index 2b4ddf4..a21d7c2 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,9 @@ RSMON_URL=https://rsmon.ru RSMON_TOKEN=replace-with-worker-token +# Required for Docker Compose. Replace with a published 64-character digest. +RSMON_WORKER_IMAGE_DIGEST=replace-with-64-lowercase-hex-digest + # Local operator console. Both credentials are required while the web UI is on. WORKER_HOST=0.0.0.0 WORKER_PORT=27401 diff --git a/README.md b/README.md index 386b01e..1a5592e 100644 --- a/README.md +++ b/README.md @@ -37,8 +37,10 @@ docker compose up -d docker compose logs -f worker ``` -Compose pulls `reg.rsxx.ru/rsmon/rsmon-worker:latest` by default. Set -`RSMON_WORKER_IMAGE` to use another published tag. +Compose constructs an immutable image reference from +`RSMON_WORKER_IMAGE_DIGEST`. Set that variable in `.env` to the published +64-character lowercase digest before running Compose; a mutable tag cannot be +selected through this configuration. The operator console is bound to `127.0.0.1:27401` by default. Set `WORKER_BIND_IP` only when a firewall or TLS reverse proxy protects the port. @@ -47,21 +49,27 @@ Persistent web and cluster state is stored in the `worker-data` volume. ## Docker ```bash -docker pull reg.rsxx.ru/rsmon/rsmon-worker:latest +IMAGE='reg.rsxx.ru/rsmon/rsmon-worker@sha256:' +docker pull "$IMAGE" docker run --rm \ --cap-add NET_RAW \ --env-file .env \ -p 127.0.0.1:27401:27401 \ -v rsmon-worker-data:/var/lib/rsmon-worker \ - reg.rsxx.ru/rsmon/rsmon-worker:latest + "$IMAGE" ``` -Published images use these tags: +Published images have tags for discovery: - `sha-<12-character-commit>` for every push; - `latest` for `master`; - the `v*` release ref, with Docker-invalid characters replaced by `-`. +Resolve a trusted published tag through the registry, then deploy the resulting +`repository@sha256:...` digest. Tags are mutable and are not accepted by the +installer or deploy command; Compose requires the resolved digest to be +configured explicitly. + The Gitea workflow reads `HARBOR_REGISTRY`, `HARBOR_USER`, and `HARBOR_PASSWORD`. `HARBOR_REGISTRY` may be a host such as `reg.rsxx.ru` or an HTTP(S) URL; the workflow strips the scheme and trailing slash before composing @@ -91,17 +99,24 @@ rm worker-token mode-0600 configuration at `/etc/rsmon-worker/worker.env`, installs a simple root-run systemd unit, and enables and starts it. Use `--url` to override `https://rsmon.ru`, `--binary` to install another binary, or `--no-start` to -configure without starting. +configure without starting. When using `--env-file`, installation verifies its +`RSMON_URL` and `RSMON_TOKEN` before changing the host. Environment files use +portable `KEY=VALUE` lines (plus blank lines and `#` comments); quoting, +interpolation, whitespace in values, and YAML-style assignments are rejected +because systemd and Docker interpret them differently. The Docker alternative pulls the prebuilt image and installs a systemd unit that runs it: ```bash -sudo ./bin/rsmon-worker install --docker --token-file worker-token +sudo ./bin/rsmon-worker install --docker --token-file worker-token \ + --image 'reg.rsxx.ru/rsmon/rsmon-worker@sha256:' ``` -The image defaults to `reg.rsxx.ru/rsmon/rsmon-worker:latest`; override it with -`--image`. +Docker installation requires `--image` with an immutable +`repository@sha256:<64-lowercase-hex-characters>` reference. This is a breaking +change: prior `--docker` invocations that relied on the `latest` default, or +passed only a tag, now fail before Docker is invoked or any host file is changed. The token can also be passed as `--token` or `--api-key`, but that can expose it through shell history and process inspection. The legacy repository-based @@ -141,7 +156,8 @@ sudo access. Add `--docker` to install remotely by uploading only the configuration and systemd unit, then running `docker pull` on the target. This mode does not upload or execute the local worker binary, so the local and remote architectures may -differ. +differ. It also requires `--image repository@sha256:...`; tag-only references +are rejected before connecting to the remote host. The default SSH port is 22 and the default RSMon URL is `https://rsmon.ru`. Encrypted keys use `--key-passphrase-file`; password authentication uses diff --git a/app/models/task_test.go b/app/models/task_test.go index 3b0cd48..62a0cea 100644 --- a/app/models/task_test.go +++ b/app/models/task_test.go @@ -251,6 +251,7 @@ func TestTasksForWorker_SkipsLockedAndLeases(t *testing.T) { for _, p := range picked { assert.Equal(t, models.TaskStateLeased, p.State) assert.Equal(t, w.WorkerID, p.LeaseOwner) + assert.NotEmpty(t, p.LeaseToken) assert.NotNil(t, p.LeaseExpiresAt) assert.Equal(t, 1, p.Attempts) } diff --git a/cmd/rsmon-worker/main.go b/cmd/rsmon-worker/main.go index e71c074..a9bc786 100644 --- a/cmd/rsmon-worker/main.go +++ b/cmd/rsmon-worker/main.go @@ -15,7 +15,6 @@ import ( "syscall" "time" - "github.com/hashicorp/raft" "github.com/joho/godotenv" "rocketgit.ru/rsmon/worker/internal/distworker" @@ -33,19 +32,6 @@ var ( // Phase 1 keeps it on; the flag exists so a Phase 2 basic-auth // install can opt out without recompiling. webappEnabled = true - - // clusterDebugApplyTestConfig, when true, submits the hardcoded - // CriticalCheckConfig from workercluster.DefaultDebugCriticalCheck - // to the cluster on startup. Wired via the - // --cluster-debug-apply-test-config CLI flag; the e2e script - // uses this so it can verify FSM replication without the - // signed-config-adoption producer (which lands in a later phase). - // - // DEBUG: this flag is a placeholder. It must be removed (or - // guarded behind a build tag) before any production build. - // - // TODO(phase-N): remove once the real producer is wired. - clusterDebugApplyTestConfig = false ) func main() { @@ -57,8 +43,6 @@ func main() { versionFlag := flag.Bool("version", false, "Print version and exit") noWeb := flag.Bool("no-web", false, "Disable the local web UI (Phase 1 ships with it on)") - debugApplyConfig := flag.Bool("cluster-debug-apply-test-config", false, - "Submit a hardcoded CriticalCheckConfig to the cluster on startup. DEBUG: remove once the real config.adopt producer is wired.") flag.Parse() if *versionFlag { @@ -66,7 +50,6 @@ func main() { os.Exit(0) } webappEnabled = !*noWeb - clusterDebugApplyTestConfig = *debugApplyConfig if len(flag.Args()) > 0 && flag.Arg(0) == "health" { os.Exit(healthCheck()) @@ -295,46 +278,12 @@ func buildCluster(ctx context.Context, cfg *distworker.Config) (*workercluster.C log.Printf("worker cluster: started node_id=%s addr=%s bootstrap=%t peers=%d", nodeID, localAddr, bootstrap, len(peers)) - if clusterDebugApplyTestConfig && bootstrap { - // The debug apply only fires on bootstrap nodes; a - // joiner cannot commit a log entry until it has been - // promoted to voter. Wait for this node to win an - // election first (a fresh single-voter cluster elects - // itself immediately but the goroutine may run before - // the state has flipped). - go func() { - deadline := time.Now().Add(15 * time.Second) - for time.Now().Before(deadline) { - if c.Raft() != nil && c.Raft().State() == raft.Leader { - break - } - time.Sleep(100 * time.Millisecond) - } - if c.Raft() == nil || c.Raft().State() != raft.Leader { - log.Printf("worker cluster: applying debug test config id=%d: not leader after 15s", - workercluster.DefaultDebugCriticalCheck().ID) - return - } - check := workercluster.DefaultDebugCriticalCheck() - applied, err := c.ApplyTestConfig(&check) - if err != nil { - log.Printf("worker cluster: applying debug test config id=%d: %v", - check.ID, err) - return - } - log.Printf("worker cluster: applying debug test config id=%d applied_index=%d", - check.ID, applied) - }() - } - return c, &clusterAdapter{c: c}, nil } // clusterAdapter wraps *workercluster.Cluster so it implements the // webapp.ClusterView interface without webapp importing the raft -// code path. The ApplyTestConfig signature is the one webapp expects -// (no config argument; the cluster subsystem owns the hardcoded -// payload so the two sides cannot drift). +// code path. type clusterAdapter struct { c *workercluster.Cluster } @@ -359,11 +308,6 @@ func (a *clusterAdapter) Stats() webapp.ClusterStats { } } -func (a *clusterAdapter) ApplyTestConfig() (uint64, error) { - check := workercluster.DefaultDebugCriticalCheck() - return a.c.ApplyTestConfig(&check) -} - func (a *clusterAdapter) ClusterID() string { return a.c.ClusterID() } func (a *clusterAdapter) LocalAddr() string { return a.c.LocalAddr() } @@ -528,13 +472,17 @@ func (w runnerWrapper) RecentNotifications(n int) []webapp.NotificationRow { out := make([]webapp.NotificationRow, len(src)) for i, r := range src { out[i] = webapp.NotificationRow{ - Kind: r.Kind, - Channel: r.Channel, - Subject: r.Subject, - Body: r.Body, - OK: r.OK, - Error: r.Error, - At: r.At, + Kind: r.Kind, + Channel: r.Channel, + Subject: r.Subject, + Body: r.Body, + OK: r.OK, + Error: r.Error, + At: r.At, + JobID: r.JobID, + Method: r.Method, + Status: r.Status, + DurationMs: r.DurationMs, } } return out diff --git a/cmd/rsmon-worker/main_test.go b/cmd/rsmon-worker/main_test.go index 00325b0..5f5c3d5 100644 --- a/cmd/rsmon-worker/main_test.go +++ b/cmd/rsmon-worker/main_test.go @@ -1,9 +1,13 @@ package main import ( + "errors" "net/http" "net/http/httptest" "os" + "os/exec" + "path/filepath" + "strings" "testing" ) @@ -52,3 +56,23 @@ func TestProbeLiveness(t *testing.T) { }) } } + +func TestRemovedClusterDebugApplyTestConfigFlagIsRejected(t *testing.T) { + binary := filepath.Join(t.TempDir(), "rsmon-worker") + build := exec.Command("go", "build", "-o", binary, ".") + if output, err := build.CombinedOutput(); err != nil { + t.Fatalf("build worker binary: %v\n%s", err, output) + } + + output, err := exec.Command(binary, "--cluster-debug-apply-test-config").CombinedOutput() + var exitErr *exec.ExitError + if !errors.As(err, &exitErr) { + t.Fatalf("removed flag error = %v, want parser exit\n%s", err, output) + } + if exitErr.ExitCode() != 2 { + t.Fatalf("removed flag exit code = %d, want 2\n%s", exitErr.ExitCode(), output) + } + if !strings.Contains(string(output), "flag provided but not defined: -cluster-debug-apply-test-config") { + t.Fatalf("removed flag output = %q, want undefined-flag error", output) + } +} diff --git a/cmd/rsmon-worker/management.go b/cmd/rsmon-worker/management.go index 6a9ce27..e8a079f 100644 --- a/cmd/rsmon-worker/management.go +++ b/cmd/rsmon-worker/management.go @@ -36,7 +36,7 @@ func installCommand(args []string) int { fs.StringVar(&tokenFile, "token-file", "", "file containing the worker API token") fs.StringVar(&opts.URL, "url", installer.DefaultURL, "RSMon server URL") fs.BoolVar(&opts.Docker, "docker", false, "run the prebuilt Docker image instead of the binary") - fs.StringVar(&opts.Image, "image", installer.DefaultImage, "Docker image used with --docker") + fs.StringVar(&opts.Image, "image", installer.DefaultImage, "immutable Docker repository@sha256 digest required with --docker") fs.BoolVar(&opts.NoStart, "no-start", false, "install and enable without starting") fs.Usage = func() { fmt.Fprintln(fs.Output(), "Usage: rsmon-worker install --token TOKEN [--url URL] [--no-start]") @@ -89,7 +89,7 @@ func deployCommand(args []string) int { fs.StringVar(&tokenFile, "token-file", "", "file containing the worker API token") fs.StringVar(&opts.URL, "url", installer.DefaultURL, "RSMon server URL") fs.BoolVar(&opts.Docker, "docker", false, "install the prebuilt Docker image remotely") - fs.StringVar(&opts.Image, "image", installer.DefaultImage, "Docker image used with --docker") + fs.StringVar(&opts.Image, "image", installer.DefaultImage, "immutable Docker repository@sha256 digest required with --docker") fs.BoolVar(&opts.NoStart, "no-start", false, "install and enable without starting") fs.Usage = func() { fmt.Fprintln(fs.Output(), "Usage: rsmon-worker deploy --host HOST --user USER --token TOKEN [SSH options]") diff --git a/docker-compose.yml b/docker-compose.yml index a3e710e..08f12a1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,6 +1,6 @@ services: worker: - image: ${RSMON_WORKER_IMAGE:-reg.rsxx.ru/rsmon/rsmon-worker:latest} + image: "reg.rsxx.ru/rsmon/rsmon-worker@sha256:${RSMON_WORKER_IMAGE_DIGEST:?set RSMON_WORKER_IMAGE_DIGEST to the published 64-character digest}" restart: unless-stopped env_file: - .env diff --git a/docs/changelog.md b/docs/changelog.md index d3749c0..67a073a 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -11,9 +11,10 @@ through `known_hosts` or a pinned fingerprint. - Added `--token-file`, `--url`, `--api-key`, and `--no-start` deployment options. -- Added optional `--docker` deployment using the prebuilt - `reg.rsxx.ru/rsmon/rsmon-worker:latest` image. Remote Docker deployment pulls - the image on the target host and does not depend on the local architecture. +- Added optional `--docker` deployment using a prebuilt image. Docker install + and deploy now require an immutable `repository@sha256:...` reference before + any Docker or remote-host mutation; the former mutable `latest` default is no + longer accepted. - Simplified the default systemd service to `Type=simple`, `User=root`, and `Restart=on-failure`. - Reworked the legacy `scripts/install-systemd.sh` script as a compatibility @@ -34,3 +35,27 @@ execution, and production worker job reporting. - Published the changes as commit `3256dcd` (`feat: add worker install and deploy`) on `master`, triggering the Docker image workflow. + +### Task protocol and local audit hardening + +- Required one task-envelope branch, matching outer/inner job IDs, and a + non-empty lease token before local execution. +- Added structured terminal failures for unsupported check kinds and safely + attributable malformed envelopes. +- Recorded delegated notification outcomes in the bounded `/notifications` + view using only job ID, method, status, duration, and time. +- Added static permanent handling for invalid deadlines and recovered + notification executor panics without retaining secret-bearing text. +- Removed the critical-cluster test-config endpoint, CLI flag, environment + switch, and production helper; hardcoded config application is test-only. +- Made runner token rotation connection-scoped and in-memory: it reconnects + without stopping web, inventory, metrics, or cluster subsystems. Durable token + storage, bootstrap exchange, rotation acknowledgement, and revocation remain + unimplemented. +- Added bounded resend of dequeued check and notification result envelopes after + websocket reconnect; control-plane application remains at-least-once and must + deduplicate by leased job and lease token. Failed metric snapshots are dropped + and replaced by the next periodic tick, not replayed. +- SIGTERM stops new dispatch and waits for active work, but stale-lease + acknowledgement, bounded graceful final-result drain, and duplicate-frame + coverage remain open. diff --git a/docs/control-plane-protocol.md b/docs/control-plane-protocol.md index bfb7efe..2d615e0 100644 --- a/docs/control-plane-protocol.md +++ b/docs/control-plane-protocol.md @@ -35,11 +35,11 @@ branch. | Server to worker | diagnostic task | new task-envelope variant | Planned | The legacy top-level `task` and `notification_task` branches are accepted for -rollout compatibility. If a frame includes both a valid current -`task_envelope` and a legacy branch, the worker executes only -`task_envelope`. The current runner ignores malformed or unsupported task -branches and does not yet compare the envelope job ID with the inner job ID; -strict rejection and reporting are P1 work below. +rollout compatibility. A current `task_envelope` must activate exactly one +matching branch with a non-empty, matching inner and outer job ID and lease +token. Unsupported check kinds and attributable malformed envelopes produce +terminal reports; malformed envelopes without a usable identity are rejected +without execution or reporting. ## Initialization And Refresh @@ -86,9 +86,10 @@ selection remains the primary isolation boundary. } ``` -`type=notification` activates `notification` instead. Job IDs in the envelope -and active branch must match once P1 validation lands. Today a task without a -recognized populated branch is ignored without execution. +`type=notification` activates `notification` instead. The worker enforces the +matching job IDs and non-empty lease token before dispatch. A task without one +recognized populated branch is rejected without execution; it is reported only +when the active branch provides attributable job and lease identity. ## Result Invariants @@ -117,20 +118,26 @@ Files: - `internal/wire/types_test.go` - `internal/distworker/runner_protocol_test.go` -Tests: +Implemented: -- decode every current frame branch; -- prefer the current envelope over duplicate legacy branches; -- reject mismatched envelope and inner job IDs; -- preserve unknown optional fields during compatible rollout; -- reject missing lease tokens before execution result submission; -- atomically replace config and credentials. +- strict single-branch envelope selection with matching job-ID and lease + validation; +- unsupported-kind and attributable malformed-task terminal reporting; +- atomic config and credential replacement. + +Still required: + +- decode conformance coverage for every current frame branch and unknown + optional fields; +- explicit drain behavior and stale-lease acknowledgement coverage. ### P2: Safe Token Rotation -Current `Runner.RotateToken` closes the runner stop channel. Replace this with a -connection-scoped cancellation path that stores the new token, closes only the -active WebSocket, and reconnects without terminating worker services. +`Runner.RotateToken` replaces the token in memory, closes only the active +control-plane WebSocket, and reconnects without terminating worker services. +The replacement is not persisted: a process restart still uses its configured +startup token. The control plane must tolerate connection-scoped result resend +using the leased job and lease token for idempotency. Acceptance: rotating from the web console produces a reconnect using the new token while the HTTP listener, collectors, and optional cluster stay running. diff --git a/docs/critical-check-cluster.md b/docs/critical-check-cluster.md index dcd8ae5..7ab7cf1 100644 --- a/docs/critical-check-cluster.md +++ b/docs/critical-check-cluster.md @@ -11,8 +11,10 @@ It does not yet execute `distributed_critical` checks, verify signed config, evaluate observation/region/notification quorum, encrypt snapshots, deliver a commit-backed outbox, or consume an external witness report. -The hardcoded test-config endpoint and startup flag are development-only and -must stay disabled in production. +The former hardcoded test-config endpoint, startup flag, environment switch, +and production helper have been removed. Hardcoded config application now +exists only as an unexported `_test.go` helper and is absent from production +builds. ## Non-Negotiable Separation @@ -163,16 +165,16 @@ but cannot commit or fabricate incidents. ## Ordered Implementation -1. Remove production exposure of debug config application. -2. Complete versioned FSM types, command validation, and deterministic tests. -3. Add mTLS identity and safe one-claim cluster bootstrap. -4. Add signed config and observer-set adoption. -5. Add deterministic scheduler and checkexec bridge in shadow mode. -6. Implement observation aggregation, incident policy, and idempotency. -7. Implement encrypted snapshots and restore/migration tests. -8. Add metadata outbox executor and failover-safe delivery. -9. Add external witness, replay, metrics, and operational runbooks. -10. Run synthetic 3/5-node fault campaigns before any customer check. +1. [x] Remove production exposure of debug config application. +2. [ ] Complete versioned FSM types, command validation, and deterministic tests. +3. [ ] Add mTLS identity and safe one-claim cluster bootstrap. +4. [ ] Add signed config and observer-set adoption. +5. [ ] Add deterministic scheduler and checkexec bridge in shadow mode. +6. [ ] Implement observation aggregation, incident policy, and idempotency. +7. [ ] Implement encrypted snapshots and restore/migration tests. +8. [ ] Add metadata outbox executor and failover-safe delivery. +9. [ ] Add external witness, replay, metrics, and operational runbooks. +10. [ ] Run synthetic 3/5-node fault campaigns before any customer check. ## Release Gates diff --git a/docs/implementation-roadmap.md b/docs/implementation-roadmap.md index f9995b9..35ab619 100644 --- a/docs/implementation-roadmap.md +++ b/docs/implementation-roadmap.md @@ -28,12 +28,15 @@ Worker repository: - wire local inventory and metrics collector lifecycle into web server start and shutdown; - validate full HTTP config, including `WORKER_URL`, in main startup; -- make token rotation reconnect rather than stop the runner; +- [x] reconnect in memory on token rotation without stopping the runner; +- [x] resend bounded check/notification results after websocket reconnect; - define process policy when the web listener exits unexpectedly; -- add WebSocket reconnect, drain, duplicate frame, and backpressure tests. +- add process-exit, bounded final-drain, duplicate-frame, and backpressure + coverage. -Gate: collectors populate real pages, rotation preserves all subsystems, and -SIGTERM leaves no listener, collector, task, or SQLite goroutine behind. +Gate: collectors populate real pages, rotation preserves all subsystems, result +resend remains bounded and idempotent, and SIGTERM leaves no listener, +collector, task, or SQLite goroutine behind. ## R2: Protocol And Credential Hardening diff --git a/docs/private-workers.md b/docs/private-workers.md index 9497f15..5f59053 100644 --- a/docs/private-workers.md +++ b/docs/private-workers.md @@ -55,19 +55,22 @@ Bootstrap replay, expiry, worker-ID mismatch, or account mismatch fails closed. ## Token Rotation -Rotation is a two-token handoff: +The current worker receives a replacement token in memory. In-memory reconnect +rotation is implemented: the control plane +invalidates the old token immediately, then `Runner.RotateToken` closes only +the active control-plane connection and reconnects with the replacement token. +It does not stop the runner or its web, inventory, metrics, or cluster +subsystems. There is no durable token write, overlap, rollback, or rotation +acknowledgement contract yet; persistent replacement-token failures can require +operator action. -1. Control plane issues a replacement token and keeps the old token valid for a - short bounded overlap. -2. Worker atomically writes the replacement secret. -3. Worker closes only the active control-plane connection and reconnects. -4. Successful authentication acknowledges rotation; control plane revokes the - old token. -5. Failure retains the old token until overlap expires and emits an operator - warning. - -The current `Runner.RotateToken` terminates the runner by closing its stop -channel. This must be fixed before claiming unattended rotation. +If a websocket write fails after the worker has dequeued a check or notification +result, it resends that envelope after reconnect. Delivery is therefore +at-least-once; the control plane must deduplicate by the leased job and lease +token before applying a resent result. Failed server-metric snapshots are +dropped because they have no idempotency key and the next periodic collection +replaces them. Metric snapshots are bound to the active control connection; +snapshots collected while disconnected or for an older connection are dropped. ## Runtime Config Authentication @@ -126,7 +129,8 @@ The standalone binary provides two systemd installation paths: installs a systemd-managed prebuilt image instead; - `rsmon-worker deploy` verifies an SSH host key, uploads the binary and a temporary mode-0600 environment file, and invokes `install` remotely; - `--docker` uploads only the environment and unit, then pulls on the target. + `--docker` uploads only the environment and unit, then pulls an explicitly + supplied immutable image digest on the target. Both default to `https://rsmon.ru` and accept `--token-file` for automation. Direct secret flags are supported but can be visible in process listings; file @@ -143,22 +147,25 @@ monitor execution. ## Implementation Work Packages -1. Add signed account/config identity to `internal/wire` and runner state. -2. Validate task account and credential scope locally before dispatch. -3. Implement one-time bootstrap and atomic token storage/rotation. -4. Add worker disable/revoke behavior and visible stale-config state. -5. Add mTLS as an optional first transport, then require it for Raft clusters. -6. Add public-task grant and SSRF-safe executor only after private isolation is +1. [ ] Add signed account/config identity to `internal/wire` and runner state. +2. [ ] Validate task account and credential scope locally before dispatch. +3. [x] Implement in-memory reconnect token rotation without stopping worker + subsystems. +4. [ ] Implement one-time bootstrap, durable token storage, and rotation + acknowledgement. +5. [ ] Add worker disable/revoke behavior and visible stale-config state. +6. [ ] Add mTLS as an optional first transport, then require it for Raft clusters. +7. [ ] Add public-task grant and SSRF-safe executor only after private isolation is proven. ## Acceptance Tests -- A private worker never leases or executes another account's normal task. -- A forged capability or account field cannot widen scope. -- Invalid, expired, downgraded, or differently scoped signed config is rejected. -- Bootstrap tokens are single-use and absent from disk after exchange. -- Rotation reconnects without stopping web, inventory, metrics, or cluster. -- Credential snapshots contain only the worker account and allowed methods. -- Public execution cannot reach loopback, link-local, RFC1918, metadata, Unix +- [ ] A private worker never leases or executes another account's normal task. +- [ ] A forged capability or account field cannot widen scope. +- [ ] Invalid, expired, downgraded, or differently scoped signed config is rejected. +- [ ] Bootstrap tokens are single-use and absent from disk after exchange. +- [x] Rotation reconnects without stopping web, inventory, metrics, or cluster. +- [ ] Credential snapshots contain only the worker account and allowed methods. +- [ ] Public execution cannot reach loopback, link-local, RFC1918, metadata, Unix sockets, or a private redirect target. -- Revocation prevents reconnect and clears in-memory credentials. +- [ ] Revocation prevents reconnect and clears in-memory credentials. diff --git a/docs/sessions/2026-07-19-standalone-deployment.md b/docs/sessions/2026-07-19-standalone-deployment.md index 361ac3d..f9a1dd7 100644 --- a/docs/sessions/2026-07-19-standalone-deployment.md +++ b/docs/sessions/2026-07-19-standalone-deployment.md @@ -29,8 +29,9 @@ default and offer the prebuilt Docker image as an explicit alternative. - The default service uses `Type=simple`, runs as `root`, and restarts only on failure. -- Docker deployment pulls `reg.rsxx.ru/rsmon/rsmon-worker:latest` on the target - host instead of uploading a local image or binary. +- Docker deployment pulls an explicitly selected immutable image digest on the + target host instead of uploading a local image or binary. The original + mutable `latest` default was superseded by a supply-chain hardening change. - The existing source-available license remains unchanged until public-release licensing is decided. - Worker credentials are supplied at deployment time and are not stored in the diff --git a/docs/tasks-and-notifications.md b/docs/tasks-and-notifications.md index d935eae..daff322 100644 --- a/docs/tasks-and-notifications.md +++ b/docs/tasks-and-notifications.md @@ -17,9 +17,11 @@ The implemented path is: Implemented check kinds are HTTP, SSL, SSH, FTP, DNS, WHOIS, BSSL, LLM, LLM-HTTP, ping, TCP, and UDP. Control-plane selection uses capabilities. The -local dispatcher supports only its explicit switch cases, but strict malformed -envelope and unsupported-kind result handling remains work package N1 rather -than a complete fail-closed protocol response. +runner and local executor share one supported-kind registry. Websocket task +envelopes require exactly one payload branch, matching non-empty outer/inner +job IDs, and a lease token. Unsupported checks return one terminal failed +result without execution. A safely attributable malformed single branch +returns a protocol failure; ambiguous and mismatched envelopes fail closed. ## Queue And Shutdown Rules @@ -29,8 +31,9 @@ than a complete fail-closed protocol response. maximum. - A task panic is recovered at the task boundary and reported as a failed attempt; it must not kill the runner. -- SIGTERM stops accepting tasks, allows bounded in-flight completion, attempts - final result delivery, and then exits. +- SIGTERM stops accepting new tasks and waits for active dispatchers before the + execution pool closes. A bounded final-result drain across websocket shutdown + is not implemented yet. - WebSocket reconnect does not re-run an in-flight or completed task. - HTTP polling remains compatibility-only and must not become a second normal scheduler. @@ -83,26 +86,27 @@ unreachable. They use system contacts and credentials from config, local jitter, and local deduplication. They must not impersonate customer delivery or mutate control-plane message state. -The local notifications page currently records selfcheck deliveries. Normal -delegated delivery attempts should be added to the same bounded view after -redaction, with job ID, method, status, duration, and time only. +The local notifications page records both selfcheck and delegated delivery +attempts. Delegated rows contain job ID, method, status, duration, and time +only. Recipient, body, credentials, hook URL, authorization, provider response, +and error text are never copied into the bounded ring. Invalid deadlines and +recovered executor panics produce static permanent results and one redacted +local row. ## Implementation Work Packages -### N1: Runner Integration Coverage +### N1: Runner Integration Coverage (partial) -Add strict envelope/inner job-ID validation and structured unsupported-kind -results. Add WebSocket integration tests for reconnect, task panic recovery, -queue backpressure, result resend, stale lease behavior, malformed envelopes, -unsupported kinds, and graceful drain. Target -`internal/distworker/runner_protocol_test.go` and a local test WebSocket server. +Strict envelope/inner job-ID/lease validation and structured unsupported-kind +results are implemented with runner-boundary tests. Live WebSocket reconnect +and result-resend tests are implemented. Stale-lease acknowledgement and +bounded graceful final-drain tests remain. -### N2: Complete Local Delivery Audit +### N2: Complete Local Delivery Audit (implemented) -Record delegated notification outcomes in the bounded local notification ring. -Never store recipient values, rendered body, credentials, or full provider -responses in SQLite. Verify `/notifications` shows both selfcheck and delegated -attempts without leaking secrets. +Delegated notification outcomes are recorded in the bounded local notification +ring. `/notifications` renders selfcheck and delegated attempts without storing +recipient values, rendered body, credentials, provider responses, or errors. ### N3: Per-Account Webhook And Mattermost Credentials diff --git a/internal/checkexec/exec.go b/internal/checkexec/exec.go index 6476673..05e3053 100644 --- a/internal/checkexec/exec.go +++ b/internal/checkexec/exec.go @@ -27,6 +27,16 @@ type ExecutedCheck struct { Metrics []wire.MetricPoint } +// SupportsKind reports whether Execute has a local executor for kind. +func SupportsKind(kind string) bool { + switch kind { + case "http", "ssl", "ssh", "ftp", "dns", "whois", "bssl", "llm", "llm-http", "ping", "tcp", "udp": + return true + default: + return false + } +} + // Execute runs checks without saving to DB or InfluxDB. // Results are returned for reporting via API to the control plane. // This is designed for distributed workers that have no direct DB access. @@ -35,6 +45,9 @@ func Execute(m *models.Monitor, checks []models.Check) []ExecutedCheck { for i := range checks { c := &checks[i] c.Monitor = m + if !SupportsKind(c.Kind) { + continue + } switch c.Kind { case "http": r := chttp.Perform(c) diff --git a/internal/checkexec/exec_test.go b/internal/checkexec/exec_test.go new file mode 100644 index 0000000..3a834bd --- /dev/null +++ b/internal/checkexec/exec_test.go @@ -0,0 +1,16 @@ +package checkexec + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestSupportsKindMatchesExecuteDispatch(t *testing.T) { + for _, kind := range []string{"http", "ssl", "ssh", "ftp", "dns", "whois", "bssl", "llm", "llm-http", "ping", "tcp", "udp"} { + assert.Truef(t, SupportsKind(kind), "kind %q", kind) + } + for _, kind := range []string{"", "rkn", "smtp"} { + assert.Falsef(t, SupportsKind(kind), "kind %q", kind) + } +} diff --git a/internal/distworker/client.go b/internal/distworker/client.go index c1d6e25..db777ff 100644 --- a/internal/distworker/client.go +++ b/internal/distworker/client.go @@ -2,12 +2,15 @@ package distworker import ( "bytes" + "context" "encoding/json" "fmt" "io" + "net" "net/http" "net/url" "strings" + "sync" "time" "github.com/gorilla/websocket" @@ -49,12 +52,16 @@ func NewClient(endpoint, authToken string) *Client { // postJSON is a helper for sending JSON POST requests func (c *Client) postJSON(path string, payload interface{}) (*http.Response, error) { + return c.postJSONContext(context.Background(), path, payload) +} + +func (c *Client) postJSONContext(ctx context.Context, path string, payload interface{}) (*http.Response, error) { body, err := json.Marshal(payload) if err != nil { return nil, err } - httpReq, err := http.NewRequest("POST", c.endpoint+path, bytes.NewReader(body)) + httpReq, err := http.NewRequestWithContext(ctx, "POST", c.endpoint+path, bytes.NewReader(body)) if err != nil { return nil, err } @@ -91,10 +98,10 @@ func (c *Client) Heartbeat(req wire.HeartbeatRequest) error { // bearer token for this worker. The current bearer is used for // authentication; the response carries the freshly issued token. // -// Returns the new token string. The main app invalidates the old -// token immediately. -func (c *Client) RotateToken() (string, error) { - resp, err := c.postJSON("/api/internal/workers/rotate-token", struct{}{}) +// Returns the new token string. The main app invalidates the old token +// immediately. +func (c *Client) RotateToken(ctx context.Context) (string, error) { + resp, err := c.postJSONContext(ctx, "/api/internal/workers/rotate-token", struct{}{}) if err != nil { return "", err } @@ -156,6 +163,12 @@ func (c *Client) ReportResults(req wire.ResultsRequest) error { // WorkerSocket connects to the websocket task channel. func (c *Client) WorkerSocket() (*websocket.Conn, error) { + return c.WorkerSocketContext(context.Background()) +} + +// WorkerSocketContext connects to the websocket task channel, cancelling the +// dial when the caller's context ends. +func (c *Client) WorkerSocketContext(ctx context.Context) (*websocket.Conn, error) { endpoint := strings.TrimRight(c.endpoint, "/") wsURL := endpoint if !strings.HasSuffix(wsURL, "/worker") && !strings.HasSuffix(wsURL, "/api/worker") { @@ -175,7 +188,33 @@ func (c *Client) WorkerSocket() (*websocket.Conn, error) { q.Set("token", c.authToken) u.RawQuery = q.Encode() - conn, resp, err := websocket.DefaultDialer.Dial(u.String(), nil) + dialer := *websocket.DefaultDialer + var ( + connMu sync.Mutex + dialConn net.Conn + ) + stopClose := context.AfterFunc(ctx, func() { + connMu.Lock() + if dialConn != nil { + _ = dialConn.Close() + } + connMu.Unlock() + }) + defer stopClose() + dialer.NetDialContext = func(dialCtx context.Context, network, address string) (net.Conn, error) { + conn, err := (&net.Dialer{}).DialContext(dialCtx, network, address) + if err != nil { + return nil, err + } + connMu.Lock() + dialConn = conn + if ctx.Err() != nil { + _ = conn.Close() + } + connMu.Unlock() + return conn, nil + } + conn, resp, err := dialer.DialContext(ctx, u.String(), nil) if err != nil && resp != nil { defer resp.Body.Close() //nolint:errcheck body, _ := io.ReadAll(resp.Body) diff --git a/internal/distworker/notification.go b/internal/distworker/notification.go index b5e9090..a4fdb08 100644 --- a/internal/distworker/notification.go +++ b/internal/distworker/notification.go @@ -25,14 +25,27 @@ import ( // silently dropping it as the production plan forbids). // //nolint:gocritic // task model is shared with the rest of the dispatcher; keep by-value -func (r *Runner) ExecuteNotification(ctx context.Context, task models.Task) wire.NotificationResultReport { +func (r *Runner) ExecuteNotification(ctx context.Context, task models.Task) (report wire.NotificationResultReport) { start := time.Now() - - report := wire.NotificationResultReport{ + method := "" + report = wire.NotificationResultReport{ JobID: task.JobID, Status: wire.NotificationResultPermanent, DurationMs: 0, } + defer func() { + if recover() != nil { + report.Status = wire.NotificationResultPermanent + report.ProviderResponse = nil + report.RetryAfterSeconds = nil + report.Error = stringPtr("notification executor panic") + } + report.DurationMs = int(time.Since(start) / time.Millisecond) + r.recordDelegatedNotification(report.JobID, method, report) + }() + if r.notificationExecutor != nil { + return r.notificationExecutor(ctx, task) + } if len(task.Payload) == 0 { report.Status = wire.NotificationResultPermanent @@ -50,6 +63,7 @@ func (r *Runner) ExecuteNotification(ctx context.Context, task models.Task) wire if nt.JobID == "" { nt.JobID = task.JobID } + method = nt.Method if nt.MessageID == 0 && task.MessageID != nil { nt.MessageID = *task.MessageID } diff --git a/internal/distworker/notification_test.go b/internal/distworker/notification_test.go index 5be7591..d5e3c8b 100644 --- a/internal/distworker/notification_test.go +++ b/internal/distworker/notification_test.go @@ -2,7 +2,10 @@ package distworker import ( "context" + "encoding/json" "errors" + "fmt" + "sync" "testing" "time" @@ -228,6 +231,177 @@ func TestExecuteNotification_ReportsJobIDFromPayload(t *testing.T) { assert.Equal(t, "outer-job", report.JobID) } +func TestDelegatedNotificationAuditCoexistsWithSelfcheckWithoutSecrets(t *testing.T) { + const ( + recipient = "ops-private@example.com" + renderedBody = "private rendered body" + credentialSecret = "smtp-password-secret" + hookURL = "https://hooks.example.com/private" + authorization = "Bearer private-authorization" + providerResponse = "private provider response" + ) + r := runnerWithCreds(&wire.NotificationCredentials{ + SMTP: []wire.SMTPCredential{{Password: credentialSecret}}, + }) + r.notifyResults = make(chan notifyResultEnvelope, 1) + r.RecordNotification(&NotificationRow{ + Kind: "email", Channel: notificationChannelSMTP, Subject: "selfcheck", Body: "main API down", OK: true, + }) + r.executeAndForwardNotification(wire.NotificationTask{ + JobID: "delegated-job", Method: "sms", Subject: authorization, + BodyText: renderedBody, BodyHTML: providerResponse, + Contact: wire.NotificationContact{Kind: "sms", Value: recipient, Name: hookURL}, + }) + + rows := r.RecentNotifications(10) + require.Len(t, rows, 2) + assert.Equal(t, "selfcheck", rows[0].Subject) + delegated := rows[1] + assert.Equal(t, "delegated-job", delegated.JobID) + assert.Equal(t, "sms", delegated.Method) + assert.Equal(t, wire.NotificationResultPermanent, delegated.Status) + assert.Empty(t, delegated.Subject) + assert.Empty(t, delegated.Body) + assert.Empty(t, delegated.Error) + assert.False(t, delegated.At.IsZero()) + + stored, err := json.Marshal(delegated) + require.NoError(t, err) + for _, secret := range []string{recipient, renderedBody, credentialSecret, hookURL, authorization, providerResponse} { + assert.NotContains(t, string(stored), secret) + } +} + +func TestDelegatedNotificationAuditRecordsOneTerminalRow(t *testing.T) { + const secret = "private-notification-secret" + validTask := func(jobID string) wire.NotificationTask { + return wire.NotificationTask{ + JobID: jobID, LeaseToken: "lease", Method: "sms", Subject: secret, BodyText: secret, + Contact: wire.NotificationContact{Value: secret, Name: secret}, + } + } + tests := []struct { + name string + invoke func(*Runner) + method string + }{ + { + name: "normal result", + invoke: func(r *Runner) { r.executeAndForwardNotification(validTask("normal")) }, + method: "sms", + }, + { + name: "invalid deadline", + invoke: func(r *Runner) { + task := validTask("invalid-deadline") + deadline := "not-a-timestamp" + task.Deadline = &deadline + r.executeAndForwardNotification(task) + }, + method: "sms", + }, + { + name: "expired deadline", + invoke: func(r *Runner) { + task := validTask("expired-deadline") + deadline := time.Now().Add(-time.Second).Format(time.RFC3339Nano) + task.Deadline = &deadline + r.executeAndForwardNotification(task) + }, + method: "sms", + }, + { + name: "malformed payload", + invoke: func(r *Runner) { + r.ExecuteNotification(context.Background(), models.Task{JobID: "malformed-payload", Payload: []byte(`{"subject":"` + secret)}) + }, + }, + { + name: "executor panic", + invoke: func(r *Runner) { + r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport { + panic(secret) + } + r.ExecuteNotification(context.Background(), models.Task{JobID: "panic", Payload: []byte(`{}`)}) + }, + }, + { + name: "malformed envelope", + invoke: func(r *Runner) { + r.enqueueTaskMessage(wire.WorkerMessage{TaskEnvelope: &wire.TaskEnvelope{ + Type: "invalid", JobID: "bad-envelope", Notify: ptrNotificationTask(validTask("bad-envelope")), + }}) + }, + method: "sms", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + r := runnerWithCreds(&wire.NotificationCredentials{}) + r.notifyResults = make(chan notifyResultEnvelope, 1) + tc.invoke(r) + + rows := r.RecentNotifications(10) + require.Len(t, rows, 1) + assert.Equal(t, wire.NotificationResultPermanent, rows[0].Status) + assert.Equal(t, tc.method, rows[0].Method) + stored, err := json.Marshal(rows[0]) + require.NoError(t, err) + assert.NotContains(t, string(stored), secret) + }) + } +} + +func TestDelegatedNotificationAuditConcurrentRowsAreSecretFree(t *testing.T) { + const attempts = 32 + const secret = "concurrent-private-secret" + r := runnerWithCreds(&wire.NotificationCredentials{}) + var wg sync.WaitGroup + for i := 0; i < attempts; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + r.ExecuteNotification(context.Background(), models.Task{ + JobID: fmt.Sprintf("job-%d", i), + Payload: []byte(`{"job_id":"job-` + fmt.Sprint(i) + `","method":"sms","body_text":"` + secret + `","contact":{"value":"` + secret + `"}}`), + }) + }(i) + } + wg.Wait() + + rows := r.RecentNotifications(attempts) + require.Len(t, rows, attempts) + for _, row := range rows { + stored, err := json.Marshal(row) + require.NoError(t, err) + assert.NotContains(t, string(stored), secret) + assert.Equal(t, wire.NotificationResultPermanent, row.Status) + } +} + +func TestInvalidNotificationDeadlineIsNotExecuted(t *testing.T) { + r := runnerWithCreds(&wire.NotificationCredentials{}) + r.notifyResults = make(chan notifyResultEnvelope, 1) + called := false + r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport { + called = true + return wire.NotificationResultReport{Status: wire.NotificationResultDelivered} + } + deadline := "not-a-timestamp" + r.executeAndForwardNotification(wire.NotificationTask{ + JobID: "invalid-deadline", LeaseToken: "lease", Method: "sms", Deadline: &deadline, + }) + + assert.False(t, called) + env := <-r.notifyResults + assert.Equal(t, wire.NotificationResultPermanent, env.report.Status) + require.NotNil(t, env.report.Error) + assert.Equal(t, "invalid notification deadline", *env.report.Error) + require.Len(t, r.RecentNotifications(10), 1) +} + +func ptrNotificationTask(task wire.NotificationTask) *wire.NotificationTask { return &task } + // guard against time import being unused if the above compile-time helpers // are dropped in a future refactor. var _ = time.Second diff --git a/internal/distworker/results.go b/internal/distworker/results.go index 3e059c6..0dfd5ee 100644 --- a/internal/distworker/results.go +++ b/internal/distworker/results.go @@ -14,10 +14,7 @@ import ( const recentResultsSize = 200 // recentNotificationsSize mirrors recentResultsSize for emitted -// notifications. Phase 1 only writes selfcheck alerts to this -// buffer (the main app's notification flow still lives in the main -// app); the buffer is shape-stable so future phases can append -// without changing the page contract. +// notifications. const recentNotificationsSize = 100 // ResultRow is one row from the worker's in-memory result ring @@ -35,9 +32,9 @@ type ResultRow struct { At time.Time } -// NotificationRow is one row from the worker's notification ring -// buffer. Phase 1 only fills this from selfcheck alerts; the row -// shape is forward-compatible with main-app-issued notifications. +// NotificationRow is one row from the worker's notification ring buffer. +// Delegated rows use only JobID, Method, Status, DurationMs, and At. They +// deliberately omit delivery inputs and provider output. type NotificationRow struct { Kind string // "email", "telegram_private", "telegram_group" Channel string @@ -46,6 +43,11 @@ type NotificationRow struct { OK bool Error string At time.Time + + JobID string + Method string + Status string + DurationMs int } // resultBuffer is a thread-safe FIFO ring buffer of ResultRow. The diff --git a/internal/distworker/runner.go b/internal/distworker/runner.go index e03fede..a123000 100644 --- a/internal/distworker/runner.go +++ b/internal/distworker/runner.go @@ -25,6 +25,8 @@ const ( // minQueueCapacity is the lower bound for the bounded job/result // channels so that a small pool still has some backpressure headroom. minQueueCapacity = 16 + + malformedTaskEnvelopeError = "malformed_task_envelope" ) // jobPool is the minimal interface the Runner needs from a worker pool. It @@ -43,6 +45,11 @@ type resultEnvelope struct { reports []wire.CheckResultReport } +type metricEnvelope struct { + generation uint64 + report wire.ServerMetricReport +} + // Runner manages the worker execution loop. // // Concurrency design: @@ -61,23 +68,32 @@ type resultEnvelope struct { // websocket writes thread-safe and removes the per-task goroutine // that previously blocked pool execution while holding the mutex. type Runner struct { - config *Config - client *Client - pool jobPool - concurrency int64 - maxConcurrency int - jobQueue chan wire.CheckJob - results chan resultEnvelope - notifyQueue chan wire.NotificationTask - notifyResults chan notifyResultEnvelope - metricResults chan wire.ServerMetricReport - stopCh chan struct{} - wg sync.WaitGroup - started atomic.Bool - queueDepth int64 - activeCount int64 - notifyDepth int64 - notifyActive int64 + config *Config + client *Client + pool jobPool + concurrency int64 + maxConcurrency int + jobQueue chan wire.CheckJob + results chan resultEnvelope + notifyQueue chan wire.NotificationTask + notifyResults chan notifyResultEnvelope + metricResults chan metricEnvelope + stopCh chan struct{} + wg sync.WaitGroup + controlWG sync.WaitGroup + backgroundWG sync.WaitGroup + lifecycleMu sync.Mutex + controlCtx context.Context + controlCancel context.CancelFunc + backgroundCtx context.Context + backgroundCancel context.CancelFunc + rotationCtx context.Context + rotationCancel context.CancelFunc + started atomic.Bool + queueDepth int64 + activeCount int64 + notifyDepth int64 + notifyActive int64 // credentialsMu guards credentials during init/config refresh. credentialsMu sync.RWMutex @@ -121,15 +137,15 @@ type Runner struct { masterStatusUp *bool masterStatusAt time.Time - // selfcheckCancel terminates the periodic selfcheck goroutine started - // by Start(). Nil until Start runs. - selfcheckCancel context.CancelFunc - // executor is the function the pool runs for each job. It is a // field so tests can swap it for a deterministic stub without // touching the websocket plumbing. executor func(payload interface{}) interface{} + // notificationExecutor is a test seam for the notification task + // boundary. Production uses ExecuteNotification's normal delivery path. + notificationExecutor func(context.Context, models.Task) wire.NotificationResultReport + // resultsBuf holds the most recent result rows. The webapp // reads from it via RecentResults(n). Cleared by Stop so the // ring does not leak between worker runs. @@ -152,26 +168,40 @@ type Runner struct { // workerID / regionCode / workerVersion / workerCaps are // captured from the most recent init/config websocket message // so the webapp can render them read-only. - workerIDMu sync.RWMutex - workerID string - regionMu sync.RWMutex - regionCode string - versionMu sync.RWMutex - workerVersion string - capsMu sync.RWMutex - workerCaps []string - serverID atomic.Int64 + workerIDMu sync.RWMutex + workerID string + regionMu sync.RWMutex + regionCode string + versionMu sync.RWMutex + workerVersion string + capsMu sync.RWMutex + workerCaps []string + serverID atomic.Int64 + metricGeneration atomic.Uint64 + nextMetricGeneration atomic.Uint64 - // clientMu guards swap of the http/websocket client during - // token rotation. The websocket loop reads r.client under the - // lock; RotateToken swaps a fresh client in under the lock - // before closing the old connection. - clientMu sync.Mutex + // clientMu guards the control-plane client and its active websocket. + // Rotation swaps the client and closes only controlConn, leaving the + // runner's global stop signal and local subsystems untouched. + clientMu sync.Mutex + controlConn *websocket.Conn + controlWriteMu *sync.Mutex + reconnectCh chan struct{} + rotationMu sync.Mutex - // closeOnce guards Close against double-close on the websocket - // from RotateToken. Phase 1 has a single websocket; RotateToken - // closes it so the reconnect loop picks up the new token. - closeOnce sync.Once + // outbox holds messages removed from a per-connection writer queue that + // could not be written before its websocket closed. + outboxMu sync.Mutex + outbox []wire.WorkerMessage + outboxWake chan struct{} + + // beforeControlWrite is a test seam used to hold a dequeued message while + // a connection rotates. + beforeControlWrite func() + + // beforeTokenCommit lets tests make shutdown win between a successful HTTP + // rotation response and the lifecycle-protected in-memory commit. + beforeTokenCommit func() } // NewRunner creates a new worker runner. Config is taken by pointer to @@ -183,10 +213,21 @@ func NewRunner(cfg *Config) *Runner { if maxConc <= 0 { maxConc = DefaultMaxConcurrency } + controlCtx, controlCancel := context.WithCancel(context.Background()) + backgroundCtx, backgroundCancel := context.WithCancel(context.Background()) + rotationCtx, rotationCancel := context.WithCancel(context.Background()) return &Runner{ config: cfg, maxConcurrency: maxConc, stopCh: make(chan struct{}), + reconnectCh: make(chan struct{}, 1), + outboxWake: make(chan struct{}, 1), + controlCtx: controlCtx, + controlCancel: controlCancel, + backgroundCtx: backgroundCtx, + backgroundCancel: backgroundCancel, + rotationCtx: rotationCtx, + rotationCancel: rotationCancel, resultsBuf: newResultBuffer(), notificationsBuf: newNotificationBuffer(), peerCache: newPeerCache(), @@ -196,23 +237,32 @@ func NewRunner(cfg *Config) *Runner { // Start begins the worker execution func (r *Runner) Start() error { log.Println("worker: starting...") + r.lifecycleMu.Lock() if r.config.URL == "" || r.config.Token == "" { + r.lifecycleMu.Unlock() return fmt.Errorf("RSMON_URL and RSMON_TOKEN must be set") } if !r.started.CompareAndSwap(false, true) { + r.lifecycleMu.Unlock() return fmt.Errorf("worker: runner already started") } + if r.stopped() { + r.lifecycleMu.Unlock() + return fmt.Errorf("worker: runner stopped") + } + r.clientMu.Lock() r.client = NewClient(r.config.URL, r.config.Token) + r.clientMu.Unlock() queueCap := r.queueCapacity() r.jobQueue = make(chan wire.CheckJob, queueCap) r.results = make(chan resultEnvelope, queueCap) r.notifyQueue = make(chan wire.NotificationTask, queueCap) r.notifyResults = make(chan notifyResultEnvelope, queueCap) - r.metricResults = make(chan wire.ServerMetricReport, queueCap) + r.metricResults = make(chan metricEnvelope, queueCap) if r.executor == nil { r.executor = r.defaultExecuteJob @@ -237,31 +287,48 @@ func (r *Runner) Start() error { go r.notifyDispatcher() } - // Start websocket task loop - go r.websocketLoop() - go r.serverMetricLoop() + // Start websocket task loop. + r.controlWG.Add(1) + go func() { + defer r.controlWG.Done() + r.websocketLoop() + }() + r.backgroundWG.Add(1) + go func() { + defer r.backgroundWG.Done() + r.serverMetricLoop(r.backgroundCtx) + }() // Start periodic selfcheck loop. This probes the main API and, on // sustained unreachability, notifies system contacts directly via // the cached credentials. Lifetimes of selfcheck goroutines are // bound to stopCh (and the explicit cancel, kept for symmetry). - selfcheckCtx, selfcheckCancel := context.WithCancel(context.Background()) - r.selfcheckCancel = selfcheckCancel - go r.startSelfcheck(selfcheckCtx) + r.backgroundWG.Add(1) + go func() { + defer r.backgroundWG.Done() + r.startSelfcheck(r.backgroundCtx) + }() // Start the peer poller. It refreshes r.peerCache with each // peer's latest /api/peer/status verdict. The selfcheck // consumes the cache to drive consensus. The poller is bound // to selfcheckCtx so it shuts down together with the // selfcheck loop on Stop. - go r.peerPollerLoop(selfcheckCtx) + r.backgroundWG.Add(1) + go func() { + defer r.backgroundWG.Done() + r.peerPollerLoop(r.backgroundCtx) + }() + r.lifecycleMu.Unlock() // Wait for stop signal <-r.stopCh log.Println("worker: shutting down...") - if r.selfcheckCancel != nil { - r.selfcheckCancel() + if r.backgroundCancel != nil { + r.backgroundCancel() } + r.controlWG.Wait() + r.backgroundWG.Wait() // Dispatchers exit via stopCh. Do not close jobQueue here: the websocket // reader can still be unwinding and may otherwise race with a send. @@ -271,21 +338,42 @@ func (r *Runner) Start() error { // otherwise an in-flight pool.Process would panic. r.pool.Close() - // Close results so any future writers exit promptly. (At this point - // the websocket connection is also gone, so this is just defensive.) - close(r.results) - return nil } // Stop gracefully stops the worker func (r *Runner) Stop() { + r.lifecycleMu.Lock() select { case <-r.stopCh: // already closed default: close(r.stopCh) } + if r.controlCancel != nil { + r.controlCancel() + } + if r.backgroundCancel != nil { + r.backgroundCancel() + } + if r.rotationCancel != nil { + r.rotationCancel() + } + r.clientMu.Lock() + conn := r.controlConn + r.clientMu.Unlock() + if conn != nil { + _ = conn.Close() + } + r.lifecycleMu.Unlock() + + // Cancel and close before waiting for rotation. A rotation can be waiting + // on controlWriteMu while a blocked writer needs that close to return. + r.rotationMu.Lock() + r.rotationMu.Unlock() + r.controlWG.Wait() + r.backgroundWG.Wait() + r.wg.Wait() } // Enqueue submits a job to the worker pool. It returns false if the runner @@ -295,6 +383,9 @@ func (r *Runner) Enqueue(job wire.CheckJob) bool { //nolint:lll,gocritic // wire if r.jobQueue == nil { return false } + if r.stopped() { + return false + } select { case r.jobQueue <- job: atomic.AddInt64(&r.queueDepth, 1) @@ -318,6 +409,9 @@ func (r *Runner) EnqueueNotification(task wire.NotificationTask) bool { if r.notifyQueue == nil { return false } + if r.stopped() { + return false + } select { case r.notifyQueue <- task: atomic.AddInt64(&r.notifyDepth, 1) @@ -350,23 +444,30 @@ func (r *Runner) notifyDispatcher() { // shape, runs the executor, and pushes the result into notifyResults. The // writer goroutine picks it up and serializes the websocket write. func (r *Runner) executeAndForwardNotification(task wire.NotificationTask) { //nolint:gocritic // wire payload is shared - if r.notifyResults == nil { - return - } - deadline := time.Now().Add(models.DefaultNotificationExecutionTimeout) + started := time.Now() + deadline := started.Add(models.DefaultNotificationExecutionTimeout) + var taskDeadline *time.Time if task.Deadline != nil { - if taskDeadline, err := time.Parse(time.RFC3339Nano, *task.Deadline); err == nil && taskDeadline.Before(deadline) { - deadline = taskDeadline + parsedDeadline, err := time.Parse(time.RFC3339Nano, *task.Deadline) + if err != nil { + r.completeNotification(task, notificationPermanentReport(task, "invalid notification deadline", started)) + return + } + taskDeadline = &parsedDeadline + if !parsedDeadline.After(started) { + r.completeNotification(task, notificationPermanentReport(task, "notification deadline expired", started)) + return + } + if parsedDeadline.Before(deadline) { + deadline = parsedDeadline } } ctx, cancel := context.WithDeadline(context.Background(), deadline) defer cancel() payload, _ := json.Marshal(task) dbTask := models.Task{JobID: task.JobID, LeaseToken: task.LeaseToken, Payload: payload} - if task.Deadline != nil { - if deadline, err := time.Parse(time.RFC3339Nano, *task.Deadline); err == nil { - dbTask.Deadline = &deadline - } + if taskDeadline != nil { + dbTask.Deadline = taskDeadline } if task.MessageID != 0 { msgID := task.MessageID @@ -377,15 +478,33 @@ func (r *Runner) executeAndForwardNotification(task wire.NotificationTask) { //n _ = ev } report := r.ExecuteNotification(ctx, dbTask) - env := notifyResultEnvelope{task: task, report: report} + r.forwardNotificationResult(task, report) +} - select { - case <-r.stopCh: +func notificationPermanentReport(task wire.NotificationTask, message string, started time.Time) wire.NotificationResultReport { + return wire.NotificationResultReport{ + JobID: task.JobID, + LeaseToken: task.LeaseToken, + MessageID: task.MessageID, + Status: wire.NotificationResultPermanent, + Error: stringPtr(message), + DurationMs: int(time.Since(started) / time.Millisecond), + } +} + +// completeNotification records the terminal local outcome before forwarding +// the result. Rejected tasks use this path because they do not enter the executor. +func (r *Runner) completeNotification(task wire.NotificationTask, report wire.NotificationResultReport) { + r.recordDelegatedNotification(report.JobID, task.Method, report) + r.forwardNotificationResult(task, report) +} + +func (r *Runner) forwardNotificationResult(task wire.NotificationTask, report wire.NotificationResultReport) { + if r.notifyResults == nil || r.stopped() { return - default: } select { - case r.notifyResults <- env: + case r.notifyResults <- notifyResultEnvelope{task: task, report: report}: case <-r.stopCh: } } @@ -482,6 +601,7 @@ func (r *Runner) websocketLoop() { select { case <-time.After(3 * time.Second): + case <-r.reconnectCh: case <-r.stopCh: return } @@ -489,29 +609,75 @@ func (r *Runner) websocketLoop() { } func (r *Runner) runWebsocket() error { - conn, err := r.client.WorkerSocket() + var writeMu sync.Mutex + r.clientMu.Lock() + client := r.client + r.clientMu.Unlock() + if client == nil { + return fmt.Errorf("worker: client not initialized") + } + + conn, err := client.WorkerSocketContext(r.controlCtx) if err != nil { return err } - defer conn.Close() //nolint:errcheck + + // A rotation can complete while the websocket dial is in flight. Do not + // install a connection authenticated with the superseded token. + r.clientMu.Lock() + if r.client != client || r.stopped() { + r.clientMu.Unlock() + _ = conn.Close() + return nil + } + r.controlConn = conn + r.controlWriteMu = &writeMu + metricGeneration := r.nextMetricGeneration.Add(1) + r.metricGeneration.Store(metricGeneration) + r.clientMu.Unlock() + defer func() { + r.clientMu.Lock() + if r.controlConn == conn { + r.controlConn = nil + r.controlWriteMu = nil + r.metricGeneration.CompareAndSwap(metricGeneration, 0) + } + r.clientMu.Unlock() + _ = conn.Close() + }() log.Println("worker: websocket connected") - var writeMu sync.Mutex done := make(chan struct{}) + var doneOnce sync.Once + closeDone := func() { doneOnce.Do(func() { close(done) }) } + var connectionWG sync.WaitGroup + defer func() { + closeDone() + _ = conn.Close() + connectionWG.Wait() + }() // Heartbeat goroutine — shares writeMu with the writer. - go r.heartbeat(conn, &writeMu, done) + connectionWG.Add(1) + go func() { + defer connectionWG.Done() + r.heartbeat(conn, &writeMu, done) + }() // Single writer goroutine for this connection: serializes result // and heartbeat writes through writeMu so websocket.WriteJSON is // never called concurrently. The dispatcher loop feeds it via the // bounded results channel. - go r.writer(conn, &writeMu, done) + connectionWG.Add(1) + go func() { + defer connectionWG.Done() + r.writer(conn, &writeMu, done, metricGeneration) + }() for { var msg wire.WorkerMessage if err := conn.ReadJSON(&msg); err != nil { - close(done) + closeDone() return err } if (msg.Kind == "init" || msg.Kind == "config") && msg.Init != nil { @@ -522,32 +688,113 @@ func (r *Runner) runWebsocket() error { continue } if !r.enqueueTaskMessage(msg) { - close(done) + closeDone() return nil } } } -// enqueueTaskMessage prefers the v2 envelope over sibling legacy fields. Some -// rollout frames contain both check representations; executing the first match -// only keeps a current runner from running one check twice. +// enqueueTaskMessage prefers the v2 envelope over sibling legacy fields. An +// envelope is accepted only when it selects exactly one matching payload with +// the same non-empty outer and inner job IDs. Invalid envelopes never fall +// back to a sibling legacy payload, which could otherwise execute a task the +// control plane did not intend to send. func (r *Runner) enqueueTaskMessage(msg wire.WorkerMessage) bool { //nolint:gocritic // wire envelope is the dispatcher boundary + if msg.TaskEnvelope != nil { + return r.enqueueTaskEnvelope(msg.TaskEnvelope) + } + switch { - case msg.TaskEnvelope != nil && msg.TaskEnvelope.Type == wire.TaskTypeNotification && msg.TaskEnvelope.Notify != nil: - return r.EnqueueNotification(*msg.TaskEnvelope.Notify) - case msg.TaskEnvelope != nil && msg.TaskEnvelope.Type == wire.TaskTypeCheck && msg.TaskEnvelope.Job != nil: - return r.Enqueue(*msg.TaskEnvelope.Job) case msg.NotificationTask != nil: + if msg.NotificationTask.JobID == "" || msg.NotificationTask.LeaseToken == "" { + return true + } log.Printf("worker: received websocket notification task %s method=%s", msg.NotificationTask.JobID, msg.NotificationTask.Method) return r.EnqueueNotification(*msg.NotificationTask) case msg.Task != nil: + if msg.Task.JobID == "" || msg.Task.LeaseToken == "" { + return true + } log.Printf("worker: received websocket task %s", msg.Task.JobID) + if !checkexec.SupportsKind(msg.Task.Kind) { + return r.enqueueUnsupportedCheck(*msg.Task) + } return r.Enqueue(*msg.Task) default: return true } } +func (r *Runner) enqueueTaskEnvelope(envelope *wire.TaskEnvelope) bool { + if (envelope.Job == nil) == (envelope.Notify == nil) { + return true + } + if envelope.Job != nil { + if !matchingEnvelopeJobID(envelope.JobID, envelope.Job.JobID) || envelope.Job.LeaseToken == "" { + return true + } + if envelope.Type != wire.TaskTypeCheck { + return r.enqueueFailedCheck(*envelope.Job, malformedTaskEnvelopeError) + } + if !checkexec.SupportsKind(envelope.Job.Kind) { + return r.enqueueUnsupportedCheck(*envelope.Job) + } + return r.Enqueue(*envelope.Job) + } + if !matchingEnvelopeJobID(envelope.JobID, envelope.Notify.JobID) || envelope.Notify.LeaseToken == "" { + return true + } + if envelope.Type != wire.TaskTypeNotification { + return r.enqueueFailedNotification(*envelope.Notify, malformedTaskEnvelopeError) + } + return r.EnqueueNotification(*envelope.Notify) +} + +func matchingEnvelopeJobID(outer, inner string) bool { + return outer != "" && outer == inner +} + +func (r *Runner) enqueueUnsupportedCheck(job wire.CheckJob) bool { + return r.enqueueFailedCheck(job, "unsupported_kind: "+job.Kind) +} + +func (r *Runner) enqueueFailedCheck(job wire.CheckJob, errorCode string) bool { + if r.results == nil || r.stopped() { + return false + } + report := wire.CheckResultReport{ + JobID: job.JobID, + CheckID: job.CheckID, + MonitorID: job.MonitorID, + State: "FAIL", + Error: stringPtr(errorCode), + DurationMs: 0, + } + select { + case r.results <- resultEnvelope{job: job, reports: []wire.CheckResultReport{report}}: + return true + case <-r.stopCh: + return false + } +} + +func (r *Runner) enqueueFailedNotification(task wire.NotificationTask, errorCode string) bool { + if r.stopped() { + return false + } + r.completeNotification(task, notificationPermanentReport(task, errorCode, time.Now())) + return true +} + +func (r *Runner) stopped() bool { + select { + case <-r.stopCh: + return true + default: + return false + } +} + func (r *Runner) heartbeat(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}) { ticker := time.NewTicker(heartbeatInterval) defer ticker.Stop() @@ -601,23 +848,39 @@ func (r *Runner) LastHeartbeatAck() time.Time { return r.lastHeartbeatAt } -func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}) { +func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}, metricGeneration uint64) { for { + if message, ok := r.takeOutbox(); ok { + if err := r.writeControlMessage(conn, writeMu, message); err != nil { + r.requeueOutbox(message) + _ = conn.Close() + return + } + continue + } + select { case env, ok := <-r.results: if !ok { return } - writeMu.Lock() for i := range env.reports { report := &env.reports[i] report.LeaseToken = env.job.LeaseToken - if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", Result: report}); err != nil { + message := wire.WorkerMessage{Kind: "result", Result: report} + if err := r.writeControlMessage(conn, writeMu, message); err != nil { + remaining := []wire.WorkerMessage{message} + for j := i + 1; j < len(env.reports); j++ { + next := env.reports[j] + next.LeaseToken = env.job.LeaseToken + remaining = append(remaining, wire.WorkerMessage{Kind: "result", Result: &next}) + } + r.requeueOutbox(remaining...) log.Printf( "worker: failed to report result job=%s check=%d kind=%s state=%s: %v", report.JobID, report.CheckID, env.job.Kind, report.State, err, ) - writeMu.Unlock() + _ = conn.Close() return } log.Printf( @@ -626,32 +889,37 @@ func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan s ) r.resultsBuf.add(resultRowFromReport(&env, report, time.Now().UTC())) } - writeMu.Unlock() case env, ok := <-r.notifyResults: if !ok { return } - writeMu.Lock() - if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", NotificationResult: &env.report}); err != nil { + message := wire.WorkerMessage{Kind: "result", NotificationResult: &env.report} + if err := r.writeControlMessage(conn, writeMu, message); err != nil { + r.requeueOutbox(message) log.Printf( "worker: failed to report notification result job=%s method=%s status=%s: %v", env.report.JobID, env.task.Method, env.report.Status, err, ) - writeMu.Unlock() + _ = conn.Close() return } log.Printf( "worker: completed notification job=%s method=%s status=%s message=%d duration_ms=%d", env.report.JobID, env.task.Method, env.report.Status, env.task.MessageID, env.report.DurationMs, ) - writeMu.Unlock() - case report := <-r.metricResults: - writeMu.Lock() - if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", ServerMetric: &report}); err != nil { - writeMu.Unlock() + case metric := <-r.metricResults: + if metric.generation != metricGeneration { + continue + } + message := wire.WorkerMessage{Kind: "result", ServerMetric: &metric.report} + if err := r.writeControlMessage(conn, writeMu, message); err != nil { + // Metrics are periodic snapshots without a lease or idempotency key. + // Dropping a failed snapshot avoids duplicate control-plane inserts; + // the next collection tick supplies a fresh replacement. + _ = conn.Close() return } - writeMu.Unlock() + case <-r.outboxWake: case <-done: return case <-r.stopCh: @@ -660,6 +928,39 @@ func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan s } } +func (r *Runner) takeOutbox() (wire.WorkerMessage, bool) { + r.outboxMu.Lock() + defer r.outboxMu.Unlock() + if len(r.outbox) == 0 { + return wire.WorkerMessage{}, false + } + message := r.outbox[0] + r.outbox = r.outbox[1:] + return message, true +} + +func (r *Runner) requeueOutbox(messages ...wire.WorkerMessage) { + if len(messages) == 0 { + return + } + r.outboxMu.Lock() + r.outbox = append(messages, r.outbox...) + r.outboxMu.Unlock() + select { + case r.outboxWake <- struct{}{}: + default: + } +} + +func (r *Runner) writeControlMessage(conn *websocket.Conn, writeMu *sync.Mutex, message wire.WorkerMessage) error { + writeMu.Lock() + defer writeMu.Unlock() + if r.beforeControlWrite != nil { + r.beforeControlWrite() + } + return conn.WriteJSON(message) +} + func (r *Runner) applyInit(init *wire.WorkerInit) { if init.Concurrency > 0 && init.Concurrency != r.Concurrency() { size := init.Concurrency @@ -881,9 +1182,6 @@ func (r *Runner) RecentResults(n int) []ResultRow { } // RecentNotifications returns the most recent n notification rows. -// Phase 1 only fills this buffer from selfcheck alerts via -// RecordNotification; the main-app-issued notifications still live -// in the main app's database. func (r *Runner) RecentNotifications(n int) []NotificationRow { if r == nil || r.notificationsBuf == nil { return nil @@ -892,8 +1190,7 @@ func (r *Runner) RecentNotifications(n int) []NotificationRow { } // RecordNotification appends one row to the notification ring buffer. -// Called from selfcheck.sendSystemAlert so the webapp /notifications -// page can show what the worker emitted. Safe before Start. +// Called from selfcheck.sendSystemAlert. Safe before Start. func (r *Runner) RecordNotification(n *NotificationRow) { if r == nil || r.notificationsBuf == nil { return @@ -904,12 +1201,24 @@ func (r *Runner) RecordNotification(n *NotificationRow) { r.notificationsBuf.add(n) } +func (r *Runner) recordDelegatedNotification(jobID, method string, report wire.NotificationResultReport) { + r.RecordNotification(&NotificationRow{ + JobID: jobID, + Method: method, + Status: report.Status, + DurationMs: report.DurationMs, + At: time.Now().UTC(), + }) +} + // Token returns the current bearer token. The webapp settings page // masks this for display. func (r *Runner) Token() string { if r == nil || r.config == nil { return "" } + r.clientMu.Lock() + defer r.clientMu.Unlock() return r.config.Token } @@ -983,48 +1292,95 @@ func (r *Runner) HTTPConfig() HTTPConfig { // // Returns the new token string. On any failure the old token and // client are kept untouched. -func (r *Runner) RotateToken(_ context.Context) (string, error) { +func (r *Runner) RotateToken(ctx context.Context) (string, error) { if r == nil || r.config == nil { return "", fmt.Errorf("worker: runner not initialized") } - if r.client == nil { + + // The endpoint authenticates with the current token, so concurrent + // rotations must not mint replacements from the same stale client. + r.rotationMu.Lock() + defer r.rotationMu.Unlock() + if r.stopped() { + return "", fmt.Errorf("worker: runner stopped") + } + requestCtx, cancelRequest := context.WithCancel(ctx) + rotationDone := make(chan struct{}) + defer func() { + close(rotationDone) + cancelRequest() + }() + go func() { + select { + case <-r.rotationCtx.Done(): + cancelRequest() + case <-rotationDone: + } + }() + + r.clientMu.Lock() + client := r.client + r.clientMu.Unlock() + if client == nil { return "", fmt.Errorf("worker: client not yet started") } - newToken, err := r.client.RotateToken() + newToken, err := client.RotateToken(requestCtx) if err != nil { return "", err } + if r.beforeTokenCommit != nil { + r.beforeTokenCommit() + } + + // Stop and the post-HTTP commit share lifecycleMu. Once Stop has closed + // stopCh, this rotation cannot install or report a replacement token. + r.lifecycleMu.Lock() + r.clientMu.Lock() + if r.stopped() { + r.clientMu.Unlock() + r.lifecycleMu.Unlock() + return "", fmt.Errorf("worker: runner stopped") + } if newToken == "" || newToken == r.config.Token { + r.clientMu.Unlock() + r.lifecycleMu.Unlock() return "", fmt.Errorf("worker: rotate-token returned unchanged or empty token") } - // Swap config + client under lock so a concurrent heartbeat - // cannot race with the rotation. - r.clientMu.Lock() + // Swap client and detach only the active control-plane connection. stopCh + // remains exclusively owned by Stop, so dispatch, metrics, selfcheck, peer + // polling, and services started alongside the runner keep running. r.config.Token = newToken - oldClient := r.client r.client = NewClient(r.config.URL, newToken) + conn := r.controlConn + writeMu := r.controlWriteMu r.clientMu.Unlock() + r.lifecycleMu.Unlock() // Stamp the rotation time so the settings page can show it. r.tokenRotatedMu.Lock() r.tokenRotatedAt = time.Now().UTC() r.tokenRotatedMu.Unlock() - // Force the websocket loop to reconnect with the new token. The - // old connection's next heartbeat will fail with 401; closing - // the connection now shortens that window. - r.closeOnce.Do(func() { - // Close the underlying websocket by triggering the runner's - // normal stop path; the websocketLoop goroutine will reconnect - // after we re-arm stopCh. This is the cleanest way to drive - // the loop without exposing internals. - select { - case <-r.stopCh: - default: - close(r.stopCh) - } - }) - _ = oldClient // client has no Close; the websocket layer owns it. + // Interrupt the current connection before waiting for its writer. A stalled + // WriteJSON holds writeMu; closing the socket makes that write fail so the + // writer can requeue its dequeued envelope and release the mutex. + if conn != nil { + _ = conn.Close() + } + if writeMu != nil { + writeMu.Lock() + writeMu.Unlock() + } + r.lifecycleMu.Lock() + stopped := r.stopped() + r.lifecycleMu.Unlock() + if stopped { + return "", fmt.Errorf("worker: runner stopped") + } + select { + case r.reconnectCh <- struct{}{}: + default: + } return newToken, nil } diff --git a/internal/distworker/runner_protocol_test.go b/internal/distworker/runner_protocol_test.go index bebaa5f..ebf290c 100644 --- a/internal/distworker/runner_protocol_test.go +++ b/internal/distworker/runner_protocol_test.go @@ -11,9 +11,162 @@ import ( func TestEnqueueTaskMessagePrefersEnvelopeOverLegacyCheck(t *testing.T) { r := &Runner{jobQueue: make(chan wire.CheckJob, 2), stopCh: make(chan struct{})} - message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{Type: wire.TaskTypeCheck, Job: &wire.CheckJob{JobID: "v2"}}, Task: &wire.CheckJob{JobID: "v1"}} + message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{Type: wire.TaskTypeCheck, JobID: "v2", Job: &wire.CheckJob{JobID: "v2", LeaseToken: "lease-2", Kind: "http"}}, Task: &wire.CheckJob{JobID: "v1"}} require.True(t, r.enqueueTaskMessage(message)) job := <-r.jobQueue assert.Equal(t, "v2", job.JobID) assert.Empty(t, r.jobQueue, "the sibling legacy field must not run a second check") } + +func TestEnqueueTaskMessageRejectsMismatchedEnvelopeJobIDs(t *testing.T) { + r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} + message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: wire.TaskTypeCheck, + JobID: "outer-job", + Job: &wire.CheckJob{JobID: "inner-job", LeaseToken: "lease-1", Kind: "http"}, + }} + + require.True(t, r.enqueueTaskMessage(message)) + assert.Empty(t, r.jobQueue) + assert.Empty(t, r.results, "a task without an unambiguous job ID cannot be reported safely") +} + +func TestEnqueueTaskMessageRejectsMissingEnvelopeJobID(t *testing.T) { + r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} + message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: wire.TaskTypeCheck, + Job: &wire.CheckJob{JobID: "inner-job", LeaseToken: "lease-1", Kind: "http"}, + }} + + require.True(t, r.enqueueTaskMessage(message)) + assert.Empty(t, r.jobQueue) + assert.Empty(t, r.results, "a missing outer ID cannot be reported safely") +} + +func TestEnqueueTaskMessageReportsMalformedCheckEnvelopeWithMatchingJobID(t *testing.T) { + r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} + message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: wire.TaskTypeNotification, + JobID: "job-1", + Job: &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", CheckID: 4, MonitorID: 5, Kind: "http"}, + }} + + require.True(t, r.enqueueTaskMessage(message)) + assert.Empty(t, r.jobQueue) + env := <-r.results + require.Len(t, env.reports, 1) + assert.Equal(t, "job-1", env.reports[0].JobID) + assert.Equal(t, "FAIL", env.reports[0].State) + require.NotNil(t, env.reports[0].Error) + assert.Equal(t, malformedTaskEnvelopeError, *env.reports[0].Error) + assert.Empty(t, r.results) +} + +func TestEnqueueTaskMessageRejectsAmbiguousEnvelope(t *testing.T) { + r := &Runner{ + jobQueue: make(chan wire.CheckJob, 1), + results: make(chan resultEnvelope, 1), + notifyQueue: make(chan wire.NotificationTask, 1), + notifyResults: make(chan notifyResultEnvelope, 1), + stopCh: make(chan struct{}), + } + message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: wire.TaskTypeCheck, + JobID: "job-1", + Job: &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", Kind: "http"}, + Notify: &wire.NotificationTask{JobID: "job-1", LeaseToken: "lease-1"}, + }} + + require.True(t, r.enqueueTaskMessage(message)) + assert.Empty(t, r.jobQueue) + assert.Empty(t, r.results) + assert.Empty(t, r.notifyQueue) + assert.Empty(t, r.notifyResults) +} + +func TestEnqueueTaskMessageReportsUnsupportedCheckKind(t *testing.T) { + r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} + message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: wire.TaskTypeCheck, + JobID: "job-1", + Job: &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", CheckID: 4, MonitorID: 5, Kind: "rkn"}, + }} + + require.True(t, r.enqueueTaskMessage(message)) + assert.Empty(t, r.jobQueue, "unsupported work must never reach the executor") + env := <-r.results + require.Len(t, env.reports, 1) + report := env.reports[0] + assert.Equal(t, "job-1", report.JobID) + assert.Equal(t, int64(4), report.CheckID) + assert.Equal(t, int64(5), report.MonitorID) + assert.Equal(t, "FAIL", report.State) + require.NotNil(t, report.Error) + assert.Equal(t, "unsupported_kind: rkn", *report.Error) + assert.Empty(t, r.results, "each rejected task must generate one terminal result") +} + +func TestEnqueueTaskMessageReportsMalformedNotificationForInvalidType(t *testing.T) { + r := &Runner{notifyQueue: make(chan wire.NotificationTask, 1), notifyResults: make(chan notifyResultEnvelope, 1), stopCh: make(chan struct{})} + message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: "unknown", + JobID: "job-1", + Notify: &wire.NotificationTask{JobID: "job-1", LeaseToken: "lease-1", MessageID: 9, Method: "email"}, + }} + + require.True(t, r.enqueueTaskMessage(message)) + assert.Empty(t, r.notifyQueue) + env := <-r.notifyResults + assert.Equal(t, "job-1", env.report.JobID) + assert.Equal(t, "lease-1", env.report.LeaseToken) + assert.Equal(t, int64(9), env.report.MessageID) + assert.Equal(t, wire.NotificationResultPermanent, env.report.Status) + require.NotNil(t, env.report.Error) + assert.Equal(t, malformedTaskEnvelopeError, *env.report.Error) + assert.Empty(t, r.notifyResults) +} + +func TestEnqueueTaskMessageRejectsEmptyLeaseWithoutSideEffects(t *testing.T) { + cases := []struct { + name string + message wire.WorkerMessage + }{ + { + name: "envelope check", + message: wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: wire.TaskTypeCheck, JobID: "job-1", Job: &wire.CheckJob{JobID: "job-1", Kind: "http"}, + }}, + }, + { + name: "envelope notification", + message: wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ + Type: wire.TaskTypeNotification, JobID: "job-1", Notify: &wire.NotificationTask{JobID: "job-1", Method: "email"}, + }}, + }, + { + name: "legacy check", + message: wire.WorkerMessage{Kind: "task", Task: &wire.CheckJob{JobID: "job-1", Kind: "http"}}, + }, + { + name: "legacy notification", + message: wire.WorkerMessage{Kind: "task", NotificationTask: &wire.NotificationTask{JobID: "job-1", Method: "email"}}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := &Runner{ + jobQueue: make(chan wire.CheckJob, 1), + results: make(chan resultEnvelope, 1), + notifyQueue: make(chan wire.NotificationTask, 1), + notifyResults: make(chan notifyResultEnvelope, 1), + stopCh: make(chan struct{}), + } + + require.True(t, r.enqueueTaskMessage(tc.message)) + assert.Empty(t, r.jobQueue) + assert.Empty(t, r.results) + assert.Empty(t, r.notifyQueue) + assert.Empty(t, r.notifyResults) + }) + } +} diff --git a/internal/distworker/runner_test.go b/internal/distworker/runner_test.go index 36f51d3..27042a0 100644 --- a/internal/distworker/runner_test.go +++ b/internal/distworker/runner_test.go @@ -1,6 +1,9 @@ package distworker import ( + "encoding/json" + "net/http" + "net/http/httptest" "strconv" "sync" "sync/atomic" @@ -8,6 +11,7 @@ import ( "time" "github.com/Jeffail/tunny" + "github.com/gorilla/websocket" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -174,6 +178,835 @@ func TestEnqueueRespectsBackpressure(t *testing.T) { } } +func TestStopRejectsTerminalResultsWithoutClosingResultChannel(t *testing.T) { + r := NewRunner(&Config{MaxConcurrency: 1}) + r.results = make(chan resultEnvelope, 1) + r.Stop() + + assert.False(t, r.enqueueFailedCheck(wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1"}, "unsupported_kind: rkn")) + assert.Empty(t, r.results) + + select { + case r.results <- resultEnvelope{}: + default: + t.Fatal("results channel should remain open after Stop") + } +} + +func TestRotateTokenReconnectsWithoutStoppingRunner(t *testing.T) { + const ( + oldToken = "old-token" + newToken = "new-token" + ) + + connections := make(chan string, 2) + results := make(chan wire.WorkerMessage, 1) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + switch req.URL.Path { + case "/api/internal/workers/rotate-token": + if req.Header.Get("Authorization") != "Bearer "+oldToken { + http.Error(w, "unexpected rotation token", http.StatusUnauthorized) + return + } + _ = json.NewEncoder(w).Encode(struct { + AuthToken string `json:"auth_token"` + }{AuthToken: newToken}) + case "/worker": + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + token := req.URL.Query().Get("token") + connections <- token + if token == oldToken { + _, _, _ = conn.ReadMessage() // Rotation must close this connection. + return + } + if token != newToken { + return + } + if conn.WriteJSON(wire.WorkerMessage{Kind: "task", Task: &wire.CheckJob{ + JobID: "after-rotation", LeaseToken: "lease", CheckID: 1, Kind: "http", + }}) != nil { + return + } + for { + var message wire.WorkerMessage + if err := conn.ReadJSON(&message); err != nil { + return + } + if message.Kind == "result" && message.Result != nil && message.Result.JobID == "after-rotation" { + results <- message + return + } + } + default: + http.NotFound(w, req) + } + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: oldToken, MaxConcurrency: 1}) + r.executor = func(payload interface{}) interface{} { + job := payload.(wire.CheckJob) + return []wire.CheckResultReport{{JobID: job.JobID, CheckID: job.CheckID, State: "OK"}} + } + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + t.Cleanup(func() { + r.Stop() + select { + case err := <-startDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("runner did not stop") + } + }) + + select { + case token := <-connections: + require.Equal(t, oldToken, token) + case <-time.After(time.Second): + t.Fatal("worker did not establish its initial control connection") + } + + gotToken, err := r.RotateToken(t.Context()) + require.NoError(t, err) + require.Equal(t, newToken, gotToken) + require.Equal(t, newToken, r.Token()) + + select { + case token := <-connections: + require.Equal(t, newToken, token) + case <-time.After(time.Second): + t.Fatal("worker did not reconnect with the replacement token") + } + select { + case result := <-results: + require.NotNil(t, result.Result) + assert.Equal(t, "OK", result.Result.State) + case <-time.After(time.Second): + t.Fatal("runner did not execute work after token rotation") + } + assert.False(t, r.stopped(), "rotation must not stop runner-owned subsystems") +} + +func TestStopClosesAndJoinsIdleControlConnection(t *testing.T) { + closed := make(chan struct{}, 1) + connected := make(chan struct{}, 1) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.URL.Path != "/worker" { + http.NotFound(w, req) + return + } + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + connected <- struct{}{} + _, _, _ = conn.ReadMessage() + closed <- struct{}{} + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1}) + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + t.Cleanup(func() { r.Stop() }) + + select { + case <-time.After(time.Second): + t.Fatal("worker did not establish idle control connection") + case <-connected: + } + r.Stop() + select { + case <-closed: + case <-time.After(time.Second): + t.Fatal("Stop did not close the idle control connection") + } + select { + case err := <-startDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("Stop did not join the control loop") + } +} + +func TestStopCancelsDialInProgress(t *testing.T) { + dialStarted := make(chan struct{}) + allowUpgrade := make(chan struct{}) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.URL.Path != "/worker" { + http.NotFound(w, req) + return + } + close(dialStarted) + <-allowUpgrade + _, _ = upgrader.Upgrade(w, req, nil) + })) + defer func() { + close(allowUpgrade) + server.Close() + }() + + r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1}) + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + select { + case <-dialStarted: + case <-time.After(time.Second): + t.Fatal("worker did not begin websocket dial") + } + r.Stop() + select { + case err := <-startDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("Stop did not join a canceled websocket dial") + } +} + +func TestStartAndStopRegisterControlLoopSafely(t *testing.T) { + server := httptest.NewServer(http.NotFoundHandler()) + defer server.Close() + for i := 0; i < 25; i++ { + r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1}) + startDone := make(chan error, 1) + stopDone := make(chan struct{}) + go func() { startDone <- r.Start() }() + go func() { + r.Stop() + close(stopDone) + }() + select { + case <-stopDone: + case <-time.After(time.Second): + t.Fatal("Stop did not complete") + } + select { + case <-startDone: + case <-time.After(time.Second): + t.Fatal("Start did not return after concurrent Stop") + } + } +} + +func TestRotateTokenPreservesDequeuedResult(t *testing.T) { + const ( + oldToken = "old-token" + newToken = "new-token" + ) + connected := make(chan string, 2) + delivered := make(chan wire.WorkerMessage, 1) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + switch req.URL.Path { + case "/api/internal/workers/rotate-token": + _ = json.NewEncoder(w).Encode(struct { + AuthToken string `json:"auth_token"` + }{AuthToken: newToken}) + case "/worker": + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + token := req.URL.Query().Get("token") + connected <- token + if token == oldToken { + var message wire.WorkerMessage + if conn.ReadJSON(&message) == nil { + delivered <- message + } + return + } + if token != newToken { + return + } + var message wire.WorkerMessage + if conn.ReadJSON(&message) == nil { + delivered <- message + } + } + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: oldToken, MaxConcurrency: 1}) + enteredWrite := make(chan struct{}) + releaseWrite := make(chan struct{}) + var once sync.Once + r.beforeControlWrite = func() { + once.Do(func() { + close(enteredWrite) + <-releaseWrite + }) + } + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + t.Cleanup(func() { + r.Stop() + select { + case <-startDone: + case <-time.After(time.Second): + t.Fatal("runner did not stop") + } + }) + + select { + case token := <-connected: + require.Equal(t, oldToken, token) + case <-time.After(time.Second): + t.Fatal("worker did not establish its initial control connection") + } + r.results <- resultEnvelope{job: wire.CheckJob{JobID: "result", LeaseToken: "lease"}, reports: []wire.CheckResultReport{{JobID: "result", State: "OK"}}} + select { + case <-enteredWrite: + case <-time.After(time.Second): + t.Fatal("writer did not dequeue result") + } + rotated := make(chan error, 1) + go func() { + _, err := r.RotateToken(t.Context()) + rotated <- err + }() + close(releaseWrite) + require.NoError(t, <-rotated) + + select { + case message := <-delivered: + require.NotNil(t, message.Result) + assert.Equal(t, "result", message.Result.JobID) + assert.Equal(t, "lease", message.Result.LeaseToken) + case <-time.After(time.Second): + t.Fatal("dequeued result was lost during rotation") + } +} + +func TestRotateTokenSerializesWithStop(t *testing.T) { + rotationStarted := make(chan struct{}) + releaseHandler := make(chan struct{}) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + switch req.URL.Path { + case "/api/internal/workers/rotate-token": + close(rotationStarted) + <-releaseHandler + case "/worker": + conn, err := upgrader.Upgrade(w, req, nil) + if err == nil { + defer conn.Close() + _, _, _ = conn.ReadMessage() + } + } + })) + defer func() { + close(releaseHandler) + server.Close() + }() + + r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1}) + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + t.Cleanup(func() { r.Stop() }) + + // Wait for Start to install its client before beginning rotation. + deadline := time.After(time.Second) + for { + r.clientMu.Lock() + started := r.client != nil + r.clientMu.Unlock() + if started { + break + } + select { + case <-deadline: + t.Fatal("runner did not start") + default: + time.Sleep(time.Millisecond) + } + } + + rotated := make(chan error, 1) + go func() { + _, err := r.RotateToken(t.Context()) + rotated <- err + }() + select { + case <-rotationStarted: + case <-time.After(time.Second): + t.Fatal("rotation request did not start") + } + stopped := make(chan struct{}) + go func() { + r.Stop() + close(stopped) + }() + require.Error(t, <-rotated, "Stop must cancel an in-flight rotation request") + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("Stop did not complete after canceling rotation") + } + select { + case err := <-startDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("runner did not stop") + } + _, err := r.RotateToken(t.Context()) + require.Error(t, err, "rotation cannot succeed after shutdown") +} + +func TestStopUnblocksRotationWaitingForWriter(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + connected := make(chan struct{}, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + switch req.URL.Path { + case "/api/internal/workers/rotate-token": + _ = json.NewEncoder(w).Encode(struct { + AuthToken string `json:"auth_token"` + }{AuthToken: "new-token"}) + case "/worker": + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + connected <- struct{}{} + _, _, _ = conn.ReadMessage() + } + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1}) + writeBlocked := make(chan struct{}) + var once sync.Once + r.beforeControlWrite = func() { + once.Do(func() { + close(writeBlocked) + <-r.controlCtx.Done() + }) + } + rotationReady := make(chan struct{}) + r.beforeTokenCommit = func() { close(rotationReady) } + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + t.Cleanup(func() { r.Stop() }) + select { + case <-connected: + case <-time.After(time.Second): + t.Fatal("worker did not connect") + } + r.results <- resultEnvelope{job: wire.CheckJob{JobID: "blocked", LeaseToken: "lease"}, reports: []wire.CheckResultReport{{JobID: "blocked", State: "OK"}}} + select { + case <-writeBlocked: + case <-time.After(time.Second): + t.Fatal("writer did not block") + } + rotated := make(chan error, 1) + go func() { + _, err := r.RotateToken(t.Context()) + rotated <- err + }() + select { + case <-rotationReady: + case <-time.After(time.Second): + t.Fatal("rotation did not reach writer serialization") + } + stopped := make(chan struct{}) + go func() { + r.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("Stop deadlocked behind rotation waiting for writer") + } + select { + case <-rotated: + case <-time.After(time.Second): + t.Fatal("rotation did not unblock after Stop closed the connection") + } + select { + case err := <-startDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("runner did not stop") + } +} + +func TestStopWinsBeforePostHTTPRotationCommit(t *testing.T) { + commitReady := make(chan struct{}) + releaseCommit := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.URL.Path != "/api/internal/workers/rotate-token" { + http.NotFound(w, req) + return + } + _ = json.NewEncoder(w).Encode(struct { + AuthToken string `json:"auth_token"` + }{AuthToken: "new-token"}) + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1}) + r.client = NewClient(server.URL, "old-token") + r.beforeTokenCommit = func() { + close(commitReady) + <-releaseCommit + } + rotated := make(chan error, 1) + go func() { + _, err := r.RotateToken(t.Context()) + rotated <- err + }() + select { + case <-commitReady: + case <-time.After(time.Second): + t.Fatal("rotation did not reach post-HTTP commit") + } + stopped := make(chan struct{}) + go func() { + r.Stop() + close(stopped) + }() + select { + case <-r.stopCh: + case <-time.After(time.Second): + t.Fatal("Stop did not win lifecycle ownership") + } + close(releaseCommit) + require.Error(t, <-rotated, "rotation cannot succeed after Stop wins") + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("Stop did not finish") + } + assert.Equal(t, "old-token", r.Token()) +} + +func TestRotateTokenUnchangedResponseReleasesLifecycle(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + _ = json.NewEncoder(w).Encode(struct { + AuthToken string `json:"auth_token"` + }{AuthToken: "old-token"}) + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1}) + r.client = NewClient(server.URL, "old-token") + _, err := r.RotateToken(t.Context()) + require.Error(t, err) + + stopped := make(chan struct{}) + go func() { + r.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("Stop deadlocked after unchanged rotation response") + } + _, err = r.RotateToken(t.Context()) + require.Error(t, err, "later rotation must observe shutdown") +} + +func TestRotateTokenClosesStalledWriterBeforeWaiting(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + connected := make(chan struct{}, 1) + connectionClosed := make(chan struct{}) + resent := make(chan wire.WorkerMessage, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + switch req.URL.Path { + case "/api/internal/workers/rotate-token": + _ = json.NewEncoder(w).Encode(struct { + AuthToken string `json:"auth_token"` + }{AuthToken: "new-token"}) + case "/worker": + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + if req.URL.Query().Get("token") == "old-token" { + connected <- struct{}{} + _, _, _ = conn.ReadMessage() + connectionClosed <- struct{}{} + return + } + var message wire.WorkerMessage + if conn.ReadJSON(&message) == nil { + resent <- message + } + } + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1}) + writerBlocked := make(chan struct{}) + var once sync.Once + r.beforeControlWrite = func() { + once.Do(func() { + close(writerBlocked) + <-connectionClosed + }) + } + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + t.Cleanup(func() { r.Stop() }) + select { + case <-connected: + case <-time.After(time.Second): + t.Fatal("worker did not connect") + } + r.results <- resultEnvelope{job: wire.CheckJob{JobID: "stalled", LeaseToken: "lease"}, reports: []wire.CheckResultReport{{JobID: "stalled", State: "OK"}}} + select { + case <-writerBlocked: + case <-time.After(time.Second): + t.Fatal("writer did not stall") + } + rotated := make(chan error, 1) + go func() { + _, err := r.RotateToken(t.Context()) + rotated <- err + }() + select { + case err := <-rotated: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("rotation waited for stalled writer before closing its connection") + } + select { + case message := <-resent: + require.NotNil(t, message.Result) + assert.Equal(t, "stalled", message.Result.JobID) + case <-time.After(time.Second): + t.Fatal("failed check result was not requeued after rotation") + } + r.Stop() + select { + case err := <-startDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("runner did not stop") + } +} + +func TestWriterDropsFailedServerMetricSnapshot(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + upgraded := make(chan struct{}) + closeServer := make(chan struct{}) + serverClosed := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + close(upgraded) + <-closeServer + close(serverClosed) + })) + defer server.Close() + + conn, err := NewClient(server.URL, "token").WorkerSocket() + require.NoError(t, err) + defer conn.Close() + select { + case <-upgraded: + case <-time.After(time.Second): + t.Fatal("websocket did not connect") + } + + r := NewRunner(&Config{}) + r.metricResults = make(chan metricEnvelope, 1) + enteredWrite := make(chan struct{}) + releaseWrite := make(chan struct{}) + r.beforeControlWrite = func() { + close(enteredWrite) + <-releaseWrite + } + done := make(chan struct{}) + writerDone := make(chan struct{}) + go func() { + var writeMu sync.Mutex + r.writer(conn, &writeMu, done, 1) + close(writerDone) + }() + r.metricResults <- metricEnvelope{generation: 1, report: wire.ServerMetricReport{ServerID: 1}} + select { + case <-enteredWrite: + case <-time.After(time.Second): + t.Fatal("writer did not dequeue metric snapshot") + } + close(closeServer) + select { + case <-serverClosed: + case <-time.After(time.Second): + t.Fatal("server did not close websocket") + } + _ = conn.Close() + close(releaseWrite) + select { + case <-writerDone: + case <-time.After(time.Second): + t.Fatal("writer did not return after failed metric write") + } + _, replayable := r.takeOutbox() + assert.False(t, replayable, "failed metric snapshot must not enter result outbox") +} + +func TestWriterRequeuesFailedNotificationResult(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + upgraded := make(chan struct{}) + closeServer := make(chan struct{}) + serverClosed := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + close(upgraded) + <-closeServer + close(serverClosed) + })) + defer server.Close() + + conn, err := NewClient(server.URL, "token").WorkerSocket() + require.NoError(t, err) + defer conn.Close() + select { + case <-upgraded: + case <-time.After(time.Second): + t.Fatal("websocket did not connect") + } + + r := NewRunner(&Config{}) + r.notifyResults = make(chan notifyResultEnvelope, 1) + enteredWrite := make(chan struct{}) + releaseWrite := make(chan struct{}) + r.beforeControlWrite = func() { + close(enteredWrite) + <-releaseWrite + } + done := make(chan struct{}) + writerDone := make(chan struct{}) + go func() { + var writeMu sync.Mutex + r.writer(conn, &writeMu, done, 0) + close(writerDone) + }() + r.notifyResults <- notifyResultEnvelope{report: wire.NotificationResultReport{JobID: "notification", LeaseToken: "lease"}} + select { + case <-enteredWrite: + case <-time.After(time.Second): + t.Fatal("writer did not dequeue notification result") + } + close(closeServer) + select { + case <-serverClosed: + case <-time.After(time.Second): + t.Fatal("server did not close websocket") + } + _ = conn.Close() + close(releaseWrite) + select { + case <-writerDone: + case <-time.After(time.Second): + t.Fatal("writer did not return after failed notification write") + } + message, replayable := r.takeOutbox() + require.True(t, replayable, "failed notification result must enter result outbox") + require.NotNil(t, message.NotificationResult) + assert.Equal(t, "notification", message.NotificationResult.JobID) +} + +func TestMetricGenerationRejectsDisconnectedAndSendsFreshMetric(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + firstClosed := make(chan struct{}) + secondConnected := make(chan struct{}) + received := make(chan wire.WorkerMessage, 1) + var connections atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + defer conn.Close() + if connections.Add(1) == 1 { + close(firstClosed) + return + } + close(secondConnected) + var message wire.WorkerMessage + if conn.ReadJSON(&message) == nil { + received <- message + } + })) + defer server.Close() + + r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1}) + startDone := make(chan error, 1) + go func() { startDone <- r.Start() }() + t.Cleanup(func() { r.Stop() }) + select { + case <-firstClosed: + case <-time.After(time.Second): + t.Fatal("worker did not establish initial websocket") + } + // Wait until the old connection has fully torn down, then try to enqueue a + // metric in the old-drain/disconnected interleaving. + deadline := time.After(time.Second) + for r.metricGeneration.Load() != 0 { + select { + case <-deadline: + t.Fatal("old control generation did not clear") + default: + time.Sleep(time.Millisecond) + } + } + staleCount := 1 + r.enqueueMetric(wire.ServerMetricReport{ServerID: 1, ProcessCount: &staleCount}) + assert.Empty(t, r.metricResults, "disconnected metric must not enter bounded channel") + select { + case r.reconnectCh <- struct{}{}: + default: + } + select { + case <-secondConnected: + case <-time.After(time.Second): + t.Fatal("worker did not reconnect") + } + deadline = time.After(time.Second) + for r.metricGeneration.Load() == 0 { + select { + case <-deadline: + t.Fatal("new control generation did not install") + default: + time.Sleep(time.Millisecond) + } + } + // This enqueue occurs immediately after the new connection installation. + freshCount := 2 + r.enqueueMetric(wire.ServerMetricReport{ServerID: 1, ProcessCount: &freshCount}) + select { + case message := <-received: + require.NotNil(t, message.ServerMetric) + assert.Equal(t, 2, *message.ServerMetric.ProcessCount) + case <-time.After(time.Second): + t.Fatal("fresh metric was not sent after reconnect") + } +} + func TestApplyInitResizesPool(t *testing.T) { executor := func(payload interface{}) interface{} { return []wire.CheckResultReport{} diff --git a/internal/distworker/server_metrics.go b/internal/distworker/server_metrics.go index dd8915c..6a1ad80 100644 --- a/internal/distworker/server_metrics.go +++ b/internal/distworker/server_metrics.go @@ -1,6 +1,7 @@ package distworker import ( + "context" "os" "path/filepath" "sort" @@ -35,32 +36,41 @@ const ( // the real Linux /proc and statfs collector; unsupported platforms return no // report rather than fabricated values. The worker has no control-plane DB // access and forwards snapshots on its authenticated websocket. -func (r *Runner) serverMetricLoop() { +func (r *Runner) serverMetricLoop(ctx context.Context) { ticker := time.NewTicker(serverMetricInterval) defer ticker.Stop() for { select { - case <-r.stopCh: + case <-ctx.Done(): return case <-ticker.C: serverID := r.serverID.Load() - if serverID == 0 || r.metricResults == nil { + if serverID == 0 { continue } report, ok := collectServerMetric(serverID) if !ok { continue } - select { - case r.metricResults <- report: - case <-r.stopCh: - return - default: - } + r.enqueueMetric(report) } } } +func (r *Runner) enqueueMetric(report wire.ServerMetricReport) { + if r.metricResults == nil { + return + } + generation := r.metricGeneration.Load() + if generation == 0 || r.metricGeneration.Load() != generation { + return + } + select { + case r.metricResults <- metricEnvelope{generation: generation, report: report}: + default: + } +} + func collectServerMetric(serverID int64) (wire.ServerMetricReport, bool) { memTotal, memAvailable, load1, load5, load15, uptime, ok := readLinuxHostMetrics("/proc") if !ok { diff --git a/internal/installer/deploy_test.go b/internal/installer/deploy_test.go index 0e79ac7..cee0bd2 100644 --- a/internal/installer/deploy_test.go +++ b/internal/installer/deploy_test.go @@ -4,6 +4,7 @@ import ( "crypto/ed25519" "crypto/rand" "net" + "strings" "testing" "golang.org/x/crypto/ssh" @@ -39,3 +40,16 @@ func TestKnownHostsMissingFile(t *testing.T) { t.Fatal("missing known_hosts file accepted") } } + +func TestDeployRejectsMutableDockerImageBeforeConnecting(t *testing.T) { + err := Deploy(DeployOptions{ + Host: "unreachable.example.test", + User: "deploy", + Token: "token", + Docker: true, + Image: "reg.rsxx.ru/rsmon/rsmon-worker:latest", + }) + if err == nil || !strings.Contains(err.Error(), "immutable") { + t.Fatalf("Deploy() error = %v, want immutable image error", err) + } +} diff --git a/internal/installer/install.go b/internal/installer/install.go index e766913..571a0cf 100644 --- a/internal/installer/install.go +++ b/internal/installer/install.go @@ -10,13 +10,17 @@ import ( "path/filepath" "regexp" "strings" + "unicode" ) -var dockerImagePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/:@-]*$`) +var ( + dockerImagePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/:-]*@sha256:[a-f0-9]{64}$`) + envKeyPattern = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`) +) const ( DefaultURL = "https://rsmon.ru" - DefaultImage = "reg.rsxx.ru/rsmon/rsmon-worker:latest" + DefaultImage = "" binaryPath = "/usr/local/bin/rsmon-worker" envPath = "/etc/rsmon-worker/worker.env" unitPath = "/etc/systemd/system/rsmon-worker.service" @@ -74,7 +78,11 @@ func Install(opts InstallOptions) error { if err := ValidateURL(opts.URL); err != nil { return err } - if opts.EnvFile == "" { + if opts.EnvFile != "" { + if err := ValidateEnvironmentFile(opts.EnvFile); err != nil { + return err + } + } else { if err := ValidateToken(opts.Token); err != nil { return err } @@ -156,9 +164,50 @@ func ValidateToken(token string) error { return nil } +// ValidateEnvironmentFile checks the worker credentials before installation +// changes the binary, systemd unit, or Docker image on the host. +func ValidateEnvironmentFile(path string) error { + data, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read worker environment: %w", err) + } + values := make(map[string]string) + seenRequired := make(map[string]bool) + for number, line := range strings.Split(string(data), "\n") { + lineNumber := number + 1 + if line == "" || strings.HasPrefix(line, "#") { + continue + } + if strings.ContainsRune(line, '\r') { + return fmt.Errorf("worker environment line %d contains a carriage return", lineNumber) + } + key, value, ok := strings.Cut(line, "=") + if !ok || !envKeyPattern.MatchString(key) { + return fmt.Errorf("worker environment line %d must use KEY=VALUE syntax", lineNumber) + } + if strings.IndexFunc(value, unicode.IsSpace) >= 0 || strings.ContainsAny(value, "$\\\"'") { + return fmt.Errorf("worker environment line %d uses unsupported quoting, interpolation, or whitespace", lineNumber) + } + if key == "RSMON_URL" || key == "RSMON_TOKEN" { + if seenRequired[key] { + return fmt.Errorf("worker environment line %d duplicates %s", lineNumber, key) + } + seenRequired[key] = true + } + values[key] = value + } + if err := ValidateURL(values["RSMON_URL"]); err != nil { + return err + } + if err := ValidateToken(values["RSMON_TOKEN"]); err != nil { + return err + } + return nil +} + func ValidateImage(image string) error { if !dockerImagePattern.MatchString(image) { - return errors.New("Docker image must be one non-option argument") + return errors.New("Docker image must be an immutable repository@sha256:<64 lowercase hex characters> reference") } return nil } diff --git a/internal/installer/install_test.go b/internal/installer/install_test.go index 5ff2934..93afeb9 100644 --- a/internal/installer/install_test.go +++ b/internal/installer/install_test.go @@ -1,6 +1,8 @@ package installer import ( + "os" + "path/filepath" "strings" "testing" ) @@ -29,6 +31,65 @@ func TestValidateToken(t *testing.T) { } } +func TestValidateEnvironmentFile(t *testing.T) { + tests := []struct { + name string + contents string + valid bool + }{ + {name: "valid", contents: "# Worker credentials\nRSMON_URL=https://rsmon.ru\nRSMON_TOKEN=secret\n\n", valid: true}, + {name: "missing URL", contents: "RSMON_TOKEN=secret\n"}, + {name: "missing token", contents: "RSMON_URL=https://rsmon.ru\n"}, + {name: "invalid URL", contents: "RSMON_URL=file:///tmp/worker\nRSMON_TOKEN=secret\n"}, + {name: "additional settings", contents: "RSMON_URL=https://rsmon.ru\nRSMON_TOKEN=secret\nWORKER_HOST=0.0.0.0\nWORKER_URL=\n", valid: true}, + {name: "dotenv interpolation", contents: "RSMON_URL=https://rsmon.ru\nRSMON_TOKEN=${TOKEN}\n"}, + {name: "dotenv export", contents: "export RSMON_URL=https://rsmon.ru\nRSMON_TOKEN=secret\n"}, + {name: "YAML assignment", contents: "RSMON_URL: https://rsmon.ru\nRSMON_TOKEN=secret\n"}, + {name: "malformed key", contents: "RSMON-URL=https://rsmon.ru\nRSMON_TOKEN=secret\n"}, + {name: "missing assignment", contents: "RSMON_URL=https://rsmon.ru\nRSMON_TOKEN\n"}, + {name: "quoted value", contents: "RSMON_URL=https://rsmon.ru\nRSMON_TOKEN=\"secret\"\n"}, + {name: "whitespace", contents: "RSMON_URL=https://rsmon.ru\nRSMON_TOKEN=secret value\n"}, + {name: "duplicate URL", contents: "RSMON_URL=https://rsmon.ru\nRSMON_URL=https://evil.test\nRSMON_TOKEN=secret\n"}, + {name: "duplicate empty token", contents: "RSMON_URL=https://rsmon.ru\nRSMON_TOKEN=\nRSMON_TOKEN=secret\n"}, + {name: "carriage return", contents: "RSMON_URL=https://rsmon.ru\r\nRSMON_TOKEN=secret\r\n"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "worker.env") + if err := os.WriteFile(path, []byte(tt.contents), 0600); err != nil { + t.Fatal(err) + } + err := ValidateEnvironmentFile(path) + if tt.valid && err != nil { + t.Fatalf("ValidateEnvironmentFile() error = %v", err) + } + if !tt.valid && err == nil { + t.Fatal("ValidateEnvironmentFile() succeeded") + } + }) + } +} + +func TestValidateEnvironmentFileInputErrors(t *testing.T) { + if err := ValidateEnvironmentFile(filepath.Join(t.TempDir(), "missing")); err == nil { + t.Fatal("missing environment file accepted") + } + if err := ValidateEnvironmentFile(t.TempDir()); err == nil { + t.Fatal("directory accepted as an environment file") + } + if os.Geteuid() == 0 { + t.Skip("root can read mode-000 files") + } + path := filepath.Join(t.TempDir(), "unreadable") + if err := os.WriteFile(path, []byte("RSMON_URL=https://rsmon.ru\nRSMON_TOKEN=secret\n"), 0000); err != nil { + t.Fatal(err) + } + if err := ValidateEnvironmentFile(path); err == nil { + t.Fatal("unreadable environment file accepted") + } +} + func TestEnvironment(t *testing.T) { got := string(Environment("https://example.test", "secret")) for _, want := range []string{"RSMON_URL=https://example.test\n", "RSMON_TOKEN=secret\n", "WORKER_HOST=127.0.0.1\n"} { @@ -45,10 +106,11 @@ func TestShellQuote(t *testing.T) { } func TestValidateImage(t *testing.T) { - if err := ValidateImage(DefaultImage); err != nil { + const image = "reg.rsxx.ru/rsmon/rsmon-worker@sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + if err := ValidateImage(image); err != nil { t.Fatal(err) } - for _, image := range []string{"", "-bad", "image name", "image%stest", `image"test`} { + for _, image := range []string{"", "-bad", "image name", "image%stest", `image"test`, "reg.rsxx.ru/rsmon/rsmon-worker:latest", "reg.rsxx.ru/rsmon/rsmon-worker@sha256:short", "reg.rsxx.ru/rsmon/rsmon-worker@sha256:0123456789ABCDEF0123456789abcdef0123456789abcdef0123456789abcdef"} { if err := ValidateImage(image); err == nil { t.Fatalf("ValidateImage(%q) succeeded", image) } @@ -59,8 +121,9 @@ func TestSystemdUnits(t *testing.T) { if !strings.Contains(systemdUnit, "Type=simple\nUser=root\n") || !strings.Contains(systemdUnit, "ExecStart=/usr/local/bin/rsmon-worker\n") { t.Fatal("binary systemd unit is not the simple root service") } - unit := DockerUnit(DefaultImage) - for _, want := range []string{"ExecStartPre=-docker rm -f rsmon-worker", "docker run --rm", DefaultImage} { + const image = "reg.rsxx.ru/rsmon/rsmon-worker@sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + unit := DockerUnit(image) + for _, want := range []string{"ExecStartPre=-docker rm -f rsmon-worker", "docker run --rm", image} { if !strings.Contains(unit, want) { t.Fatalf("Docker systemd unit missing %q", want) } diff --git a/internal/webapp/constants.go b/internal/webapp/constants.go index 58d945d..8b541d7 100644 --- a/internal/webapp/constants.go +++ b/internal/webapp/constants.go @@ -24,14 +24,13 @@ const ( // Environment variable names referenced by ConfigFromEnv. Lifted out // so the validator and the cmd binary share the same constants. const ( - envWorkerHost = "WORKER_HOST" - envWorkerPort = "WORKER_PORT" - envWorkerURL = "WORKER_URL" - envWorkerLogin = "WORKER_LOGIN" - envWorkerPassword = "WORKER_PASSWORD" - envClusterEnabled = "WORKER_CLUSTER_ENABLED" - envClusterDebugApply = "WORKER_CLUSTER_DEBUG_APPLY" - envReleaseURL = "WORKER_RELEASE_URL" + envWorkerHost = "WORKER_HOST" + envWorkerPort = "WORKER_PORT" + envWorkerURL = "WORKER_URL" + envWorkerLogin = "WORKER_LOGIN" + envWorkerPassword = "WORKER_PASSWORD" + envClusterEnabled = "WORKER_CLUSTER_ENABLED" + envReleaseURL = "WORKER_RELEASE_URL" ) // Route paths used as redirect targets. Lifted out so goconst stops diff --git a/internal/webapp/handlers_cluster.go b/internal/webapp/handlers_cluster.go index f4e34e6..2668af4 100644 --- a/internal/webapp/handlers_cluster.go +++ b/internal/webapp/handlers_cluster.go @@ -74,63 +74,6 @@ func (s *Server) handleClusterStatus(w http.ResponseWriter, _ *http.Request) { } } -// handleClusterApplyTestConfig applies a hardcoded config.adopt log -// entry to the cluster. It exists so the e2e script and any operator -// debugging session can verify FSM replication without having to wire -// up the real signed-config-adoption producer (which lives in a later -// phase). -// -// DEBUG: this endpoint is a placeholder for the real producer. It must -// be replaced (or removed) before any production deployment. -// -// The handler is gated behind Config.DebugClusterApply (env -// WORKER_CLUSTER_DEBUG_APPLY=true). When the flag is false the -// handler returns 404 — the route is still registered so the auth -// + CSRF paths are exercised in tests, but no real FSM entry is ever -// appended from a production webapp. -// -// TODO(worker-cluster-real-producer): remove the apply-test-config -// endpoint entirely once the signed-config-adoption producer ships. -func (s *Server) handleClusterApplyTestConfig(w http.ResponseWriter, r *http.Request) { - writeNoStore(w) - if !s.cfg.DebugClusterApply { - http.NotFound(w, r) - return - } - if s.cluster == nil { - http.Error(w, "cluster not configured", http.StatusServiceUnavailable) - return - } - if !s.requireCSRF(sessionFromContextOrFail(w, r), r) { - http.Error(w, "csrf token required", http.StatusForbidden) - return - } - applied, err := s.cluster.ApplyTestConfig() - if err != nil { - s.deps.Logger.Printf("cluster apply test config: %v", err) - http.Error(w, err.Error(), http.StatusBadGateway) - return - } - w.Header().Set("Content-Type", "application/json; charset=utf-8") - w.WriteHeader(http.StatusOK) - if err := json.NewEncoder(w).Encode(map[string]uint64{"applied_index": applied}); err != nil { - s.deps.Logger.Printf("cluster apply encode: %v", err) - } -} - -// sessionFromContextOrFail is a tiny adapter so requireCSRF can be -// called from this handler without leaking the middleware into the -// cluster package. If no session is attached (should not happen -// because requireSession already ran) we return a stub session with -// no CSRF token, which causes requireCSRF to refuse the request. -func sessionFromContextOrFail(_ http.ResponseWriter, r *http.Request) *Session { - sess, _ := sessionFromContext(r.Context()) - if sess != nil { - return sess - } - return &Session{} -} - // ErrClusterNotConfigured is returned when a cluster-admin endpoint is // hit on a server without a cluster attached. var ErrClusterNotConfigured = errors.New("webapp: cluster not configured") diff --git a/internal/webapp/handlers_cluster_test.go b/internal/webapp/handlers_cluster_test.go index c20648f..76f5a06 100644 --- a/internal/webapp/handlers_cluster_test.go +++ b/internal/webapp/handlers_cluster_test.go @@ -1,16 +1,11 @@ package webapp import ( - "context" "encoding/json" "io" "net/http" "net/http/httptest" - "net/url" - "strings" - "sync" "testing" - "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -21,23 +16,13 @@ import ( // pinned without standing up a real raft group. type stubCluster struct { stats ClusterStats - applyIndex uint64 - applyErr error - applyCalled int - applyMu sync.Mutex clusterIDOut string addrOut string } func (s *stubCluster) Stats() ClusterStats { return s.stats } -func (s *stubCluster) ApplyTestConfig() (uint64, error) { - s.applyMu.Lock() - defer s.applyMu.Unlock() - s.applyCalled++ - return s.applyIndex, s.applyErr -} -func (s *stubCluster) ClusterID() string { return s.clusterIDOut } -func (s *stubCluster) LocalAddr() string { return s.addrOut } +func (s *stubCluster) ClusterID() string { return s.clusterIDOut } +func (s *stubCluster) LocalAddr() string { return s.addrOut } // withClusterServer returns a test server whose ClusterView is the // supplied stub. The first-run password path is also exercised so @@ -166,151 +151,23 @@ func TestClusterStatus_RequiresSession(t *testing.T) { assert.Equal(t, "/web/login", resp.Header.Get("Location")) } -// TestClusterApplyTestConfig_NotConfigured verifies the 404 path -// when WORKER_CLUSTER_DEBUG_APPLY is false (the production default) -// and no cluster is attached. The handler must refuse before it -// even checks the cluster because the debug flag is off. -func TestClusterApplyTestConfig_NotConfigured(t *testing.T) { - srv := newTestServer(t, &stubRunner{id: "w-1"}) - require.False(t, srv.cfg.DebugClusterApply, "default config must leave the debug apply flag off") - ts := newHTTPTestServer(t, srv) - c, _ := loginAsFirstRun(t, ts.URL, srv) - clearRequiresChange(t, srv) +// TestClusterApplyTestConfig_NotExposed verifies production requests cannot +// append a hardcoded config through the former debug endpoint. +func TestClusterApplyTestConfig_NotExposed(t *testing.T) { + t.Setenv("WORKER_CLUSTER_DEBUG_APPLY", "true") + ts, _, c := withClusterServer(t, &stubCluster{}) - resp, err := c.PostForm(ts.URL+"/web/api/cluster/apply-test-config", url.Values{}) - require.NoError(t, err) - defer resp.Body.Close() //nolint:errcheck - assert.Equal(t, http.StatusNotFound, resp.StatusCode, - "debug apply must be invisible (404) when WORKER_CLUSTER_DEBUG_APPLY is unset") -} - -// TestClusterApplyTestConfig_DebugOffReturns404 verifies that even -// with a cluster attached the apply endpoint stays 404 unless the -// debug flag is on. The flag, not cluster presence, gates the -// endpoint. -func TestClusterApplyTestConfig_DebugOffReturns404(t *testing.T) { - stub := &stubCluster{applyIndex: 42} - ts, _, c := withClusterServer(t, stub) - - resp, err := c.Get(ts.URL + "/overview") - require.NoError(t, err) - bodyBytes, _ := io.ReadAll(resp.Body) - resp.Body.Close() //nolint:errcheck - csrf := extractCSRFToken(t, string(bodyBytes)) - - form := url.Values{} - form.Set("csrf_token", csrf) - req, _ := http.NewRequest(http.MethodPost, ts.URL+"/web/api/cluster/apply-test-config", - strings.NewReader(form.Encode())) - req.Header.Set("Content-Type", "application/x-www-form-urlencoded") - resp, err = c.Do(req) + resp, err := c.Post(ts.URL+"/web/api/cluster/apply-test-config", "", nil) require.NoError(t, err) defer resp.Body.Close() //nolint:errcheck assert.Equal(t, http.StatusNotFound, resp.StatusCode) - assert.Equal(t, 0, stub.applyCalled, - "ApplyTestConfig must never be called when the debug flag is off") } -// TestClusterApplyTestConfig_HappyPath verifies that the apply-test- -// config endpoint returns the applied index when the cluster -// subsystem accepts the entry. CSRF is checked. The DebugClusterApply -// flag must be on for the endpoint to be reachable. -func TestClusterApplyTestConfig_HappyPath(t *testing.T) { - stub := &stubCluster{ - stats: ClusterStats{ - NodeID: "worker1", State: "Leader", Leader: "worker1", - Voters: []string{"worker1"}, - }, - applyIndex: 13, - clusterIDOut: "worker1", - addrOut: "127.0.0.1:17401", - } - ts, srv, c := withClusterServer(t, stub) - srv.cfg.DebugClusterApply = true - - // Fetch CSRF token from any authenticated page. - resp, err := c.Get(ts.URL + "/overview") - require.NoError(t, err) - bodyBytes, _ := io.ReadAll(resp.Body) - resp.Body.Close() //nolint:errcheck - csrf := extractCSRFToken(t, string(bodyBytes)) - - form := url.Values{} - form.Set("csrf_token", csrf) - req, _ := http.NewRequest(http.MethodPost, ts.URL+"/web/api/cluster/apply-test-config", - strings.NewReader(form.Encode())) - req.Header.Set("Content-Type", "application/x-www-form-urlencoded") - resp, err = c.Do(req) - require.NoError(t, err) - defer resp.Body.Close() //nolint:errcheck - assert.Equal(t, http.StatusOK, resp.StatusCode) - bodyBytes, _ = io.ReadAll(resp.Body) - - var got map[string]uint64 - require.NoError(t, json.Unmarshal(bodyBytes, &got)) - assert.EqualValues(t, 13, got["applied_index"]) - - assert.Equal(t, 1, stub.applyCalled) -} - -// TestClusterApplyTestConfig_PropagatesError verifies that errors -// from the cluster subsystem surface as 502 Bad Gateway. Debug flag -// must be on. -func TestClusterApplyTestConfig_PropagatesError(t *testing.T) { - stub := &stubCluster{ - applyErr: errStubApply, - clusterIDOut: "worker1", - addrOut: "127.0.0.1:17401", - } - ts, srv, c := withClusterServer(t, stub) - srv.cfg.DebugClusterApply = true - - resp, err := c.Get(ts.URL + "/overview") - require.NoError(t, err) - bodyBytes, _ := io.ReadAll(resp.Body) - resp.Body.Close() //nolint:errcheck - csrf := extractCSRFToken(t, string(bodyBytes)) - - form := url.Values{} - form.Set("csrf_token", csrf) - req, _ := http.NewRequest(http.MethodPost, ts.URL+"/web/api/cluster/apply-test-config", - strings.NewReader(form.Encode())) - req.Header.Set("Content-Type", "application/x-www-form-urlencoded") - resp, err = c.Do(req) - require.NoError(t, err) - defer resp.Body.Close() //nolint:errcheck - assert.Equal(t, http.StatusBadGateway, resp.StatusCode) -} - -// TestClusterApplyTestConfig_RequiresCSRF ensures the apply-test- -// config POST is refused without a CSRF token. Debug flag must be -// on for the endpoint to be reachable; without the flag it returns -// 404 (priority over CSRF check). -func TestClusterApplyTestConfig_RequiresCSRF(t *testing.T) { - stub := &stubCluster{applyIndex: 99} - ts, srv, c := withClusterServer(t, stub) - srv.cfg.DebugClusterApply = true - - resp, err := c.PostForm(ts.URL+"/web/api/cluster/apply-test-config", url.Values{}) - require.NoError(t, err) - defer resp.Body.Close() //nolint:errcheck - assert.Equal(t, http.StatusForbidden, resp.StatusCode, - "apply-test-config without CSRF must be 403") - assert.Equal(t, 0, stub.applyCalled, "ApplyTestConfig must not be called without CSRF") -} - -// errStubApply is a sentinel error used by the apply-error test. -var errStubApply = errApply("worker not leader") - -type errApply string - -func (e errApply) Error() string { return string(e) } - // TestSetClusterDetaches verifies SetCluster(nil) returns the server // to the no-cluster-attached state (503 from the endpoints). func TestSetClusterDetaches(t *testing.T) { srv := newTestServer(t, &stubRunner{id: "w-1"}) - stub := &stubCluster{applyIndex: 7} + stub := &stubCluster{} srv.SetCluster(stub) require.NotNil(t, srv.Cluster()) @@ -326,11 +183,3 @@ func TestSetClusterDetaches(t *testing.T) { defer resp.Body.Close() //nolint:errcheck assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) } - -// _ = context.Background and time.Time keep the linter quiet about -// unused imports if the file shrinks. -var ( - _ = context.Background - _ = time.Now - _ = url.Parse -) diff --git a/internal/webapp/handlers_updates_test.go b/internal/webapp/handlers_updates_test.go index 5b80c5e..46f6369 100644 --- a/internal/webapp/handlers_updates_test.go +++ b/internal/webapp/handlers_updates_test.go @@ -224,7 +224,10 @@ func TestChecksPage_RunNowDisabledButton(t *testing.T) { // TestNotificationsPage_ResendDisabledButton mirrors the checks // page test for the resend button on /notifications. func TestNotificationsPage_ResendDisabledButton(t *testing.T) { - srv := newTestServer(t, &stubRunner{id: "w-1"}) + srv := newTestServer(t, &stubRunner{id: "w-1", notifs: []NotificationRow{ + {Kind: "email", Channel: "smtp", Subject: "selfcheck", Body: "main API down", OK: true, At: time.Now()}, + {JobID: "delegated-job", Method: "sms", Status: "permanent", DurationMs: 12, At: time.Now()}, + }}) ts := newHTTPTestServer(t, srv) c, _ := loginAsFirstRun(t, ts.URL, srv) clearRequiresChange(t, srv) @@ -239,6 +242,11 @@ func TestNotificationsPage_ResendDisabledButton(t *testing.T) { assert.Contains(t, page, "Resend") assert.Contains(t, page, "disabled") assert.Contains(t, page, "worker-notifier-mvp") + assert.Contains(t, page, "selfcheck") + assert.Contains(t, page, "delegated-job") + assert.Contains(t, page, "sms") + assert.Contains(t, page, "permanent") + assert.Contains(t, page, "12 ms") } // TestAppsPage_ReferencesInventoryPlan ensures the copy on diff --git a/internal/webapp/routes.go b/internal/webapp/routes.go index f9ff0da..d68bd31 100644 --- a/internal/webapp/routes.go +++ b/internal/webapp/routes.go @@ -62,7 +62,6 @@ func (s *Server) routes() { // Both routes require a session (the worker webapp is single-tenant // so every logged-in operator is effectively an admin). s.mux.Handle("GET /web/api/cluster/status", s.requireSession(s.handleClusterStatus)) - s.mux.Handle("POST /web/api/cluster/apply-test-config", s.requireSession(s.handleClusterApplyTestConfig)) // Cross-worker peer status. The path is intentionally under // /api/ (not /web/api/) so the basic-auth middleware does not diff --git a/internal/webapp/server.go b/internal/webapp/server.go index 716f0e2..93d3a6d 100644 --- a/internal/webapp/server.go +++ b/internal/webapp/server.go @@ -80,16 +80,6 @@ type Config struct { BasicAuthLogin string BasicAuthPassword string - // DebugClusterApply gates the /web/api/cluster/apply-test-config - // endpoint. When false (the default) the route is registered but - // the handler returns 404 so the endpoint is invisible in - // production. Operators who want to poke the cluster FSM during - // development set WORKER_CLUSTER_DEBUG_APPLY=true. The endpoint - // must NEVER be reachable in production — it appends hardcoded - // log entries to the Raft FSM without going through the real - // config-adoption producer. - DebugClusterApply bool - // ReleaseURL is the optional URL the worker polls to discover // the latest published version of the worker binary. When empty // the /updates page shows the placeholder "v1 (dev)". The URL @@ -186,7 +176,6 @@ func ConfigFromEnv(env map[string]string, defaultDataDir string) (Config, error) if v := env["RSMON_WEBAPP_STORE_PATH"]; v != "" { cfg.StorePath = v } - cfg.DebugClusterApply = parseBool(env[envClusterDebugApply]) cfg.ReleaseURL = strings.TrimSpace(env[envReleaseURL]) return cfg, nil } @@ -212,7 +201,7 @@ func ConfigFromEnvOrDefault() Config { envWorkerHost, envWorkerPort, envWorkerURL, envWorkerLogin, envWorkerPassword, "RSMON_WEBAPP_DATA_DIR", "RSMON_WEBAPP_STORE_PATH", "WORKER_CLUSTER_ENABLED", - envClusterDebugApply, envReleaseURL, + envReleaseURL, } { if v := os.Getenv(k); v != "" { env[k] = v @@ -254,7 +243,6 @@ type Deps struct { // pulling in the raft package or bbolt. type ClusterView interface { Stats() ClusterStats - ApplyTestConfig() (uint64, error) ClusterID() string LocalAddr() string } @@ -320,9 +308,7 @@ type ResultRow struct { At time.Time } -// NotificationRow is one row from the worker's in-memory notification -// ring buffer. Phase 1 only emits selfcheck alerts; main-app-issued -// notifications still live in the main app's DB. +// NotificationRow is one row from the worker's in-memory notification ring. type NotificationRow struct { Kind string // "email", "telegram_private", "telegram_group" Channel string @@ -331,6 +317,11 @@ type NotificationRow struct { OK bool Error string At time.Time + + JobID string + Method string + Status string + DurationMs int } // Server is the local HTTP server for the worker webapp. It owns the diff --git a/internal/webapp/server_test.go b/internal/webapp/server_test.go index 740c9dc..b282db1 100644 --- a/internal/webapp/server_test.go +++ b/internal/webapp/server_test.go @@ -88,40 +88,6 @@ func TestConfigFromEnvBasicAuthRejectsXOR(t *testing.T) { assert.Error(t, err, "XOR (password only) must be rejected") } -// TestConfigFromEnvDebugClusterApply pins the default-off behavior -// of the cluster-apply debug gate and verifies the env flag flips -// it on. Production builds must not accidentally expose the -// endpoint, so the default is false. -func TestConfigFromEnvDebugClusterApply(t *testing.T) { - cfg, err := ConfigFromEnv(map[string]string{}, t.TempDir()) - require.NoError(t, err) - assert.False(t, cfg.DebugClusterApply, "default must leave the debug flag off") - - cfg, err = ConfigFromEnv(map[string]string{ - "WORKER_CLUSTER_DEBUG_APPLY": "true", - }, t.TempDir()) - require.NoError(t, err) - assert.True(t, cfg.DebugClusterApply) - - // Other truthy spellings accepted. - for _, v := range []string{"yes", "1", "TRUE", "YeS"} { - cfg, err = ConfigFromEnv(map[string]string{ - "WORKER_CLUSTER_DEBUG_APPLY": v, - }, t.TempDir()) - require.NoError(t, err) - assert.True(t, cfg.DebugClusterApply, "must accept truthy value %q", v) - } - - // Empty / unknown values stay false. - for _, v := range []string{"", "false", "0", "no"} { - cfg, err = ConfigFromEnv(map[string]string{ - "WORKER_CLUSTER_DEBUG_APPLY": v, - }, t.TempDir()) - require.NoError(t, err) - assert.False(t, cfg.DebugClusterApply, "must reject non-truthy value %q", v) - } -} - // TestConfigFromEnvReleaseURL pins the env-driven WORKER_RELEASE_URL // plumbing. The handler reads cfg.ReleaseURL when the page renders, // so the value must survive ConfigFromEnv exactly. diff --git a/internal/webapp/templates/notifications.html b/internal/webapp/templates/notifications.html index 9297ca3..5655715 100644 --- a/internal/webapp/templates/notifications.html +++ b/internal/webapp/templates/notifications.html @@ -1,15 +1,15 @@ {{define "body"}}

Recent notifications

-

Last 50 notifications emitted by this worker (Phase 1: selfcheck alerts only).

+

Last 50 notification attempts emitted by this worker.

{{.ResendTooltip}}

- - - + + + @@ -17,15 +17,23 @@ {{range .Rows}} - - - + {{if .JobID}} + + + + + + {{else}} + + + + {{end}} {{else}} - + {{end}}
ChannelSubjectBodyTypeDetails StatusDuration Error When
{{.Channel}} ({{.Kind}}){{.Subject}}{{.Body}}delegatedjob {{.JobID}}, {{.Method}}{{.Status}}{{.DurationMs}} msselfcheck{{.Channel}} ({{.Kind}}): {{.Subject}} {{.Body}} {{if .OK}}delivered{{else}}failed{{end}}- {{.Error}}{{fmtTime .At}}
No notifications yet.
No notifications yet.
diff --git a/internal/workercluster/admin_test.go b/internal/workercluster/admin_test.go index 90facc3..602f71d 100644 --- a/internal/workercluster/admin_test.go +++ b/internal/workercluster/admin_test.go @@ -72,11 +72,9 @@ func TestClusterSmokeStats(t *testing.T) { assert.Equal(t, addr, c.LocalAddr()) } -// TestClusterApplyTestConfig_Smoke verifies the ApplyTestConfig helper -// commits a config.adopt entry and the FSM reflects the new version. -// This is the function the webapp admin endpoint and the CLI flag -// both go through. -func TestClusterApplyTestConfig_Smoke(t *testing.T) { +// TestClusterTestConfig_Smoke verifies the test-only helper commits a +// config.adopt entry and the FSM reflects the new version. +func TestClusterTestConfig_Smoke(t *testing.T) { addr := pickPort(t) dataDir := t.TempDir() creds := HTTPCreds{Login: "alice", Password: "secret"} @@ -108,8 +106,8 @@ func TestClusterApplyTestConfig_Smoke(t *testing.T) { require.Equal(t, raft.Leader, c.Raft().State(), "smoke test requires the local node to be leader") - check := DefaultDebugCriticalCheck() - idx, err := c.ApplyTestConfig(&check) + check := defaultTestCriticalCheck() + idx, err := c.applyTestConfig(&check) require.NoError(t, err) assert.NotZero(t, idx, "applied index must be non-zero") @@ -164,11 +162,11 @@ func TestClusterStats_FSMFieldsOnFreshCluster(t *testing.T) { "newly created FSM defaults Partition.State to 'steady'") } -// TestApplyTestConfig_NonLeaderErrors pins the precondition that the +// TestTestConfig_NonLeaderErrors pins the precondition that the // helper refuses to submit an entry on a non-leader (the raft library // would reject the apply anyway, but we want the failure to be // deterministic and informative). -func TestApplyTestConfig_NonLeaderErrors(t *testing.T) { +func TestTestConfig_NonLeaderErrors(t *testing.T) { // Three-node fixture so we have a clear "not the leader" node. if testing.Short() { t.Skip("3-node smoke skipped in -short mode") @@ -187,9 +185,9 @@ func TestApplyTestConfig_NonLeaderErrors(t *testing.T) { require.NotNil(t, leader) require.NotNil(t, follower) - check := DefaultDebugCriticalCheck() - _, err := follower.ApplyTestConfig(&check) - require.Error(t, err, "non-leader must refuse ApplyTestConfig") + check := defaultTestCriticalCheck() + _, err := follower.applyTestConfig(&check) + require.Error(t, err, "non-leader must refuse test config application") assert.Contains(t, strings.ToLower(err.Error()), "not leader") } diff --git a/internal/workercluster/cluster.go b/internal/workercluster/cluster.go index 1a7fe7f..f5ce1c5 100644 --- a/internal/workercluster/cluster.go +++ b/internal/workercluster/cluster.go @@ -2,7 +2,6 @@ package workercluster import ( "context" - "encoding/json" "errors" "fmt" "io" @@ -325,63 +324,6 @@ func (c *Cluster) Snapshot() error { return r.Snapshot().Error() } -// DefaultDebugCriticalCheck returns the hardcoded CriticalCheckConfig -// the cluster admin debug endpoint and the -// --cluster-debug-apply-test-config CLI flag apply. A fresh Epoch is -// stamped on every call so repeated applies produce distinct entries -// (handy for verifying replication timing). -func DefaultDebugCriticalCheck() CriticalCheckConfig { - return CriticalCheckConfig{ - ID: 9999, - Kind: "distributed_critical", - IntervalS: 30, - Target: "http://example.com", - Epoch: time.Now().UTC().UnixNano(), - } -} - -// ApplyTestConfig submits a hardcoded config.adopt log entry with the -// supplied CriticalCheckConfig. Returns the applied log index. This -// is a debug convenience used by the e2e script and the -// --cluster-debug-apply-test-config CLI flag; production code should -// build entries from the real signed-config-adoption producer -// (Phase-N work). -// -// DEBUG: this exists only so the e2e shell script can verify FSM -// replication without a real producer wired in. -// -// TODO(phase-N): remove once the real producer lands. -func (c *Cluster) ApplyTestConfig(check *CriticalCheckConfig) (uint64, error) { - c.mu.Lock() - r := c.raft - c.mu.Unlock() - if r == nil { - return 0, errors.New("workercluster: not started") - } - if r.State() != raft.Leader { - return 0, errors.New("workercluster: not leader; submit on the leader") - } - - payload := ConfigAdoptPayload{ - Version: 1, - Actor: c.opts.NodeID, - Checks: []CriticalCheckConfig{*check}, - } - raw, err := json.Marshal(payload) - if err != nil { - return 0, fmt.Errorf("workercluster: encode payload: %w", err) - } - entry, err := EncodeEntry(&Entry{Type: EntryConfigAdopt, Adopted: raw}) - if err != nil { - return 0, fmt.Errorf("workercluster: encode entry: %w", err) - } - fut := r.Apply(entry, 10*time.Second) - if err := fut.Error(); err != nil { - return 0, fmt.Errorf("workercluster: apply test config: %w", err) - } - return fut.Index(), nil -} - // ClusterID returns the NodeID this cluster was constructed with. It // is exposed so HTTP handlers can label status responses with a // stable identifier even when the raft library's own State() reports diff --git a/internal/workercluster/test_config_test.go b/internal/workercluster/test_config_test.go new file mode 100644 index 0000000..f92a72c --- /dev/null +++ b/internal/workercluster/test_config_test.go @@ -0,0 +1,49 @@ +package workercluster + +import ( + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/hashicorp/raft" +) + +// defaultTestCriticalCheck is test-only scaffold input for replication tests. +func defaultTestCriticalCheck() CriticalCheckConfig { + return CriticalCheckConfig{ + ID: 9999, + Kind: "distributed_critical", + IntervalS: 30, + Target: "http://example.com", + Epoch: time.Now().UTC().UnixNano(), + } +} + +// applyTestConfig is deliberately compiled only into workercluster tests. +func (c *Cluster) applyTestConfig(check *CriticalCheckConfig) (uint64, error) { + c.mu.Lock() + r := c.raft + c.mu.Unlock() + if r == nil { + return 0, errors.New("workercluster: not started") + } + if r.State() != raft.Leader { + return 0, errors.New("workercluster: not leader; submit on the leader") + } + + payload := ConfigAdoptPayload{Version: 1, Actor: c.opts.NodeID, Checks: []CriticalCheckConfig{*check}} + raw, err := json.Marshal(payload) + if err != nil { + return 0, fmt.Errorf("workercluster: encode payload: %w", err) + } + entry, err := EncodeEntry(&Entry{Type: EntryConfigAdopt, Adopted: raw}) + if err != nil { + return 0, fmt.Errorf("workercluster: encode entry: %w", err) + } + fut := r.Apply(entry, 10*time.Second) + if err := fut.Error(); err != nil { + return 0, fmt.Errorf("workercluster: apply test config: %w", err) + } + return fut.Index(), nil +}