diff --git a/README.md b/README.md index 6c737b0..36cb1b4 100644 --- a/README.md +++ b/README.md @@ -70,10 +70,12 @@ The raw auth key is valid only while it matches the single current encrypted key In Docker, `RUN_PLATFORM_URL` must be `http://platform:8080` because `platform` is the compose service name. Locally, keep it as `http://127.0.0.1:8080`. -Current executable behavior includes smoke mode plus worker mode. Worker mode registers with platform, sends lightweight heartbeat metadata, claims lifecycle jobs, acknowledges leases, reports bounded progress, executes scoped `process.install`, `process.start`, and `process.stop` command templates inside per-server workspaces, polls cancellation, submits terminal results, and reconciles active jobs. +Current executable behavior includes smoke mode plus worker mode. Worker mode registers with platform, opens a signed persistent control event stream for lightweight wakeups, sends heartbeat metadata, claims lifecycle jobs, acknowledges leases, reports bounded progress, executes scoped `process.install`, `process.start`, and `process.stop` command templates inside per-server workspaces, polls cancellation, submits terminal results, and reconciles active jobs. Lifecycle templates are JSON files addressed by logical keys under the server workspace. They resolve to direct executable/argument vectors, not shell strings. Absolute paths, parent traversal, raw credentials, direct sockets, shell launchers, unsafe environment keys, and unsafe output are rejected or redacted. Plugin-declared Windows `.cmd` and `.bat` assets are launched through a bounded `cmd.exe` adapter and remain under the same process supervisor. Process identity journals are namespaced by Run endpoint, server, plugin, and component profile, so multiple Run services cannot overwrite one another; a restart migrates matching legacy state, reopens the persisted output files, reconciles the PID, and resumes stdout/stderr tailing. Process stdout/stderr is written to the log spool, and lifecycle result metadata is queued through artifact hooks so control heartbeat and job result submission stay independent from log and artifact work. +The control event stream carries only small hints such as `control.ready`, `control.heartbeat`, and `job.changed`. It never carries assignments, logs, artifact chunks, file bodies, host paths, credentials, or direct sockets; Run still fetches work through the durable job claim channel after a wake event. + ## Runtime Profiles And Distribution Jobs Run resolves plugin-declared runtime profiles using server runtime bindings supplied by platform. Supported modes are: diff --git a/api/platform_client.go b/api/platform_client.go index 8404f7f..986e653 100644 --- a/api/platform_client.go +++ b/api/platform_client.go @@ -1,6 +1,7 @@ package api import ( + "bufio" "bytes" "context" "crypto/hmac" @@ -148,6 +149,84 @@ func (c PlatformClient) Heartbeat(ctx context.Context, request protocol.RunHeart return postPlatformJSON[protocol.RunHeartbeatRequest, protocol.RunHeartbeatResponse](ctx, c, "/api/v1/run/control/heartbeat", request) } +func (c PlatformClient) StreamControlEvents(ctx context.Context, request protocol.RunControlStreamRequest, handle func(protocol.RunControlEvent) error) error { + if handle == nil { + return fmt.Errorf("control event handler is required") + } + startedAt := time.Now() + path := "/api/v1/run/control/events" + log.Printf("RUN platform stream status=starting method=POST base=%s path=%s", diagnosticLogValue(c.baseURL), path) + var body bytes.Buffer + if err := json.NewEncoder(&body).Encode(request); err != nil { + return fmt.Errorf("encode control stream request: %w", err) + } + httpRequest, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, &body) + if err != nil { + return fmt.Errorf("build control stream request: %w", err) + } + httpRequest.Header.Set("Content-Type", "application/json") + httpRequest.Header.Set("Accept", "text/event-stream") + signatureSummary, err := signRunRequest(httpRequest, body.Bytes()) + if err != nil { + return err + } + log.Printf("RUN platform stream status=signed method=POST base=%s path=%s endpoint=%s timestamp=%s nonce=%s bodyHash=%s signature=%s", diagnosticLogValue(c.baseURL), path, diagnosticLogValue(signatureSummary.RunEndpointID), signatureSummary.Timestamp, shortDiagnosticValue(signatureSummary.Nonce), shortDiagnosticValue(signatureSummary.BodyHash), shortDiagnosticValue(signatureSummary.Signature)) + httpResponse, err := c.httpClient.Do(httpRequest) + if err != nil { + log.Printf("RUN platform stream status=send_error method=POST base=%s path=%s durationMs=%d error=%s", diagnosticLogValue(c.baseURL), path, time.Since(startedAt).Milliseconds(), err) + return fmt.Errorf("send control stream request: %w", err) + } + defer httpResponse.Body.Close() + log.Printf("RUN platform stream status=response method=POST base=%s path=%s httpStatus=%d durationMs=%d", diagnosticLogValue(c.baseURL), path, httpResponse.StatusCode, time.Since(startedAt).Milliseconds()) + if httpResponse.StatusCode < http.StatusOK || httpResponse.StatusCode >= http.StatusMultipleChoices { + var failure struct { + Code string `json:"code"` + Details []string `json:"details"` + } + _ = json.NewDecoder(io.LimitReader(httpResponse.Body, 64<<10)).Decode(&failure) + return PlatformRequestError{Status: httpResponse.StatusCode, Path: path, Code: failure.Code, Details: failure.Details} + } + reader := bufio.NewReader(httpResponse.Body) + var eventName string + var dataLines []string + for { + line, err := reader.ReadString('\n') + if err != nil && len(line) == 0 { + if errors.Is(err, io.EOF) || ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("read control stream: %w", err) + } + line = strings.TrimRight(line, "\r\n") + if line == "" { + if len(dataLines) > 0 { + var event protocol.RunControlEvent + if err := json.Unmarshal([]byte(strings.Join(dataLines, "\n")), &event); err != nil { + return fmt.Errorf("decode control event: %w", err) + } + if event.Type == "" { + event.Type = eventName + } + if err := handle(event); err != nil { + return err + } + } + eventName = "" + dataLines = nil + } else if strings.HasPrefix(line, "event:") { + eventName = strings.TrimSpace(strings.TrimPrefix(line, "event:")) + } else if strings.HasPrefix(line, "data:") { + dataLines = append(dataLines, strings.TrimSpace(strings.TrimPrefix(line, "data:"))) + } + if err != nil { + if errors.Is(err, io.EOF) || ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("read control stream: %w", err) + } + } +} + func (c PlatformClient) ReportLifecycle(ctx context.Context, request protocol.RunLifecycleReportRequest) (protocol.RunLifecycleReportResponse, error) { return postPlatformJSON[protocol.RunLifecycleReportRequest, protocol.RunLifecycleReportResponse](ctx, c, "/api/v1/run/lifecycle/report", request) } diff --git a/api/platform_client_test.go b/api/platform_client_test.go index e9de91e..87cb334 100644 --- a/api/platform_client_test.go +++ b/api/platform_client_test.go @@ -136,6 +136,45 @@ func TestPlatformClientSignsRunChannelRequestsWithUniqueNonce(t *testing.T) { } } +func TestPlatformClientStreamsSignedControlEvents(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/api/v1/run/control/events" { + t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path) + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatalf("read stream body: %v", err) + } + verifyRunRequestSignature(t, r, body, "run-local", "session-token") + var request protocol.RunControlStreamRequest + if err := json.Unmarshal(body, &request); err != nil { + t.Fatalf("decode stream request: %v", err) + } + if request.LastEventSeq != 7 { + t.Fatalf("unexpected stream cursor: %+v", request) + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("event: job.changed\nid: 8\ndata: {\"runEndpointId\":\"run-local\",\"sequence\":8,\"type\":\"job.changed\",\"serverTime\":\"2026-07-03T12:00:00Z\"}\n\n")) + })) + defer server.Close() + + client, err := NewPlatformClient(server.URL) + if err != nil { + t.Fatalf("new client: %v", err) + } + events := []protocol.RunControlEvent{} + err = client.StreamControlEvents(context.Background(), protocol.RunControlStreamRequest{RunEndpointID: "run-local", SessionToken: "session-token", LastEventSeq: 7}, func(event protocol.RunControlEvent) error { + events = append(events, event) + return nil + }) + if err != nil { + t.Fatalf("stream events: %v", err) + } + if len(events) != 1 || events[0].Type != protocol.RunControlEventTypeJobChanged || events[0].Sequence != 8 { + t.Fatalf("unexpected control events: %+v", events) + } +} + func verifyRunRequestSignature(t *testing.T, request *http.Request, body []byte, endpoint string, token string) { t.Helper() if request.Header.Get("X-Run-Endpoint") != endpoint { diff --git a/protocol/control.go b/protocol/control.go index 832e79f..b3e8406 100644 --- a/protocol/control.go +++ b/protocol/control.go @@ -2,6 +2,12 @@ package protocol import "time" +const ( + RunControlEventTypeReady = "control.ready" + RunControlEventTypeHeartbeat = "control.heartbeat" + RunControlEventTypeJobChanged = "job.changed" +) + type RunCapacityReport struct { MaxJobs int `json:"maxJobs"` RunningJobs int `json:"runningJobs"` @@ -60,6 +66,20 @@ type RunHeartbeatResponse struct { ServerTime time.Time `json:"serverTime"` } +type RunControlStreamRequest struct { + RunEndpointID string `json:"runEndpointId"` + SessionToken string `json:"sessionToken"` + LastEventSeq uint64 `json:"lastEventSeq,omitempty"` +} + +type RunControlEvent struct { + RunEndpointID string `json:"runEndpointId"` + Sequence uint64 `json:"sequence"` + Type string `json:"type"` + ServerTime time.Time `json:"serverTime"` + RetrySeconds int `json:"retrySeconds,omitempty"` +} + type RunLifecycleReportRequest struct { RunEndpointID string `json:"runEndpointId"` SessionToken string `json:"sessionToken"` diff --git a/protocol/control.md b/protocol/control.md index 850096b..2aa8c1f 100644 --- a/protocol/control.md +++ b/protocol/control.md @@ -6,6 +6,7 @@ Control is the lightweight high-priority channel between run and platform. - `POST /api/v1/run/control/hello`: registers run metadata and receives a platform-issued session token. - `POST /api/v1/run/control/heartbeat`: reports status, capacity, and capability fingerprint using the active session token. +- `POST /api/v1/run/control/events`: opens a signed `text/event-stream` control channel. Platform sends small wake events such as `control.ready`, `control.heartbeat`, and `job.changed`; Run then claims durable jobs through the job channel. ## Payloads @@ -13,12 +14,15 @@ Control is the lightweight high-priority channel between run and platform. - `RunHelloResponse`: session token, server time, polling hints, and feature flags. - `RunHeartbeatRequest`: session token, version, status, capacity, and current capability fingerprint. - `RunHeartbeatResponse`: accepted status, next heartbeat interval, and optional capability refresh request. +- `RunControlStreamRequest`: run ID, session token, and the last event sequence observed by Run. +- `RunControlEvent`: run ID, monotonic endpoint-local sequence, event type, server time, and reconnect hint. - `RunCapabilityReport`: capability names and compact fingerprint metadata. - `RunCapacityReport`: max jobs, active jobs, queued jobs, and local resource summary. ## Rules - Control payloads must be small. +- Control events wake Run only; they must not carry job assignments or execution input. - Control must not carry logs, artifact chunks, or long job result bodies. - Control must have priority over job execution, log upload, and artifact transfer. - Heartbeat capacity summaries must remain metadata-only and must not mention or carry heavy channel payloads. diff --git a/protocol/job.md b/protocol/job.md index 4298b2f..9192ba1 100644 --- a/protocol/job.md +++ b/protocol/job.md @@ -4,7 +4,7 @@ Jobs execute bounded server management work. ## Implemented Routes -- `POST /api/v1/run/jobs/claim`: claims one queued job for the registered run endpoint, optionally holding the request for a bounded wait window so Platform can wake Run immediately when work arrives. +- `POST /api/v1/run/jobs/claim`: claims one queued job for the registered run endpoint. New workers are normally awakened by the persistent control event stream and claim without holding this request; older workers may still use a bounded `waitSeconds` long-poll fallback. - `POST /api/v1/run/jobs/ack`: acknowledges an active leased job before execution. - `POST /api/v1/run/jobs/progress`: reports bounded progress for an active leased job. - `POST /api/v1/run/jobs/result`: submits a bounded terminal result for an active leased job. @@ -17,7 +17,7 @@ Jobs execute bounded server management work. ## Payloads -- `RunJobClaimRequest`: session token, run ID, capacity, supported capabilities, and optional `waitSeconds` for long-poll claim waiting. +- `RunJobClaimRequest`: session token, run ID, capacity, supported capabilities, and optional `waitSeconds` for legacy long-poll claim waiting. - `RunJobClaimResponse`: optional job assignment with identity, capability, server instance, logical target key, scoped input ref, idempotency key, per-job attempt, max attempts, raw one-use lease token, ack deadline, execution lease deadline, and polling hint. Platform persists only the lease hash. - `RunJobAckRequest`: job ID, run ID, session token, lease token, attempt, and bounded message. - `RunJobProgressRequest`: job ID, run ID, session token, lease token, attempt, percent, sequence, and bounded message. @@ -78,6 +78,7 @@ The executor resolves lifecycle action templates under the scoped server workspa - Job payloads must not include logs, artifact chunks, raw host paths, raw credentials, direct sockets, or large inline result bodies. - Process stdout/stderr must be redacted and written to the log spool rather than embedded in progress/result bodies. - Job ack/progress/result/cancel/reconcile calls are lightweight lifecycle metadata and must be able to complete while artifact/file transfer work is active or retrying. +- Persistent control events are hints only. Run must still claim, ack, execute, and complete durable jobs through this job channel so retries, leases, and idempotency stay platform-authoritative. - Dependency adapters and update downloads run in the job worker while control heartbeat, cancellation polling, log spool upload, and artifact upload retain independent bounded loops. - Terminal results must remain idempotent under log and artifact retry pressure and must reference artifacts by safe `artifact://...` refs rather than embedding transfer payloads. diff --git a/runtime/worker.go b/runtime/worker.go index 48c17de..23a8228 100644 --- a/runtime/worker.go +++ b/runtime/worker.go @@ -24,6 +24,7 @@ import ( type WorkerClient interface { Hello(context.Context, protocol.RunHelloRequest) (protocol.RunHelloResponse, error) Heartbeat(context.Context, protocol.RunHeartbeatRequest) (protocol.RunHeartbeatResponse, error) + StreamControlEvents(context.Context, protocol.RunControlStreamRequest, func(protocol.RunControlEvent) error) error ReportLifecycle(context.Context, protocol.RunLifecycleReportRequest) (protocol.RunLifecycleReportResponse, error) GetRunLogStreamProgress(context.Context, protocol.RunLogStreamProgressRequest) (protocol.RunLogStreamProgressResponse, error) ClaimJob(context.Context, protocol.RunJobClaimRequest) (protocol.RunJobClaimResponse, error) @@ -430,6 +431,10 @@ func (worker *Worker) reregisterAndReconcile(ctx context.Context, reason string, } func (worker *Worker) ClaimAndRunOnce(ctx context.Context) (bool, error) { + return worker.claimAndRunOnce(ctx, jobClaimWaitSeconds) +} + +func (worker *Worker) claimAndRunOnce(ctx context.Context, waitSeconds int) (bool, error) { state, err := worker.registeredState() if err != nil { return false, err @@ -440,7 +445,7 @@ func (worker *Worker) ClaimAndRunOnce(ctx context.Context) (bool, error) { SessionToken: state.SessionToken, Capabilities: state.Capabilities, Capacity: worker.capacityReportFor(state), - WaitSeconds: jobClaimWaitSeconds, + WaitSeconds: waitSeconds, }) if err != nil { if sessionInvalidError(err) { @@ -973,7 +978,10 @@ func (worker *Worker) Run(ctx context.Context) error { uploaderDone := make(chan struct{}) go worker.runDurableUploaders(workerCtx, uploaderDone) jobDone := make(chan error, 1) - go func() { jobDone <- worker.runJobLoop(workerCtx, jobInterval) }() + controlWake := make(chan struct{}, 1) + controlDone := make(chan error, 1) + go func() { controlDone <- worker.runControlStreamLoop(workerCtx, controlWake) }() + go func() { jobDone <- worker.runJobLoop(workerCtx, jobInterval, controlWake) }() defer func() { cancelWorker() <-uploaderDone @@ -986,6 +994,12 @@ func (worker *Worker) Run(ctx context.Context) error { case err := <-jobDone: log.Printf("RUN phase=run status=job_loop_done error=%s", errorSummary(err)) return err + case err := <-controlDone: + if err != nil && err != context.Canceled { + log.Printf("RUN phase=run status=control_stream_done error=%s", errorSummary(err)) + return err + } + return err case <-heartbeatTicker.C: if err := worker.HeartbeatOnce(ctx); err != nil { log.Printf("RUN phase=heartbeat status=retry_scheduled backoffMs=%d", boundedRetryBackoff(worker.cfg.RetryBackoff).Milliseconds()) @@ -1036,8 +1050,59 @@ func (worker *Worker) reportRunUpdateHealth(ctx context.Context) error { return nil } -func (worker *Worker) runJobLoop(ctx context.Context, interval time.Duration) error { - log.Printf("RUN phase=job_loop status=starting fallbackPollMs=%d claimWaitSeconds=%d", interval.Milliseconds(), jobClaimWaitSeconds) +func (worker *Worker) runControlStreamLoop(ctx context.Context, wake chan<- struct{}) error { + log.Printf("RUN phase=control_stream status=starting endpoint=%s", worker.cfg.RunEndpointID) + var lastSeq uint64 + for { + state, err := worker.registeredState() + if err != nil { + if waitErr := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); waitErr != nil { + return waitErr + } + continue + } + err = worker.client.StreamControlEvents(ctx, protocol.RunControlStreamRequest{RunEndpointID: state.RunEndpointID, SessionToken: state.SessionToken, LastEventSeq: lastSeq}, func(event protocol.RunControlEvent) error { + if event.RunEndpointID != "" && event.RunEndpointID != state.RunEndpointID { + return fmt.Errorf("control event endpoint mismatch") + } + if event.Sequence > lastSeq { + lastSeq = event.Sequence + } + log.Printf("RUN phase=control_stream status=event endpoint=%s type=%s seq=%d", state.RunEndpointID, safeOptional(event.Type), event.Sequence) + signalControlWake(wake) + return nil + }) + if ctx.Err() != nil { + log.Printf("RUN phase=control_stream status=context_done error=%s", RedactText(ctx.Err().Error())) + return ctx.Err() + } + if err != nil { + if sessionInvalidError(err) { + log.Printf("RUN phase=control_stream status=session_invalid endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + if refreshErr := worker.reregisterAndReconcile(ctx, "control_stream_session_invalid", state.SessionToken); refreshErr != nil { + log.Printf("RUN phase=control_stream status=reregister_failed endpoint=%s error=%s", state.RunEndpointID, RedactText(refreshErr.Error())) + } + lastSeq = 0 + signalControlWake(wake) + } else { + log.Printf("RUN phase=control_stream status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + } + } + if waitErr := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); waitErr != nil { + return waitErr + } + } +} + +func signalControlWake(wake chan<- struct{}) { + select { + case wake <- struct{}{}: + default: + } +} + +func (worker *Worker) runJobLoop(ctx context.Context, interval time.Duration, wake <-chan struct{}) error { + log.Printf("RUN phase=job_loop status=starting fallbackPollMs=%d controlStream=true", interval.Milliseconds()) for { select { case <-ctx.Done(): @@ -1063,7 +1128,7 @@ func (worker *Worker) runJobLoop(ctx context.Context, interval time.Duration) er } } claimStartedAt := time.Now() - handled, err := worker.ClaimAndRunOnce(ctx) + handled, err := worker.claimAndRunOnce(ctx, 0) if err != nil { log.Printf("RUN phase=job_loop status=claim_failed error=%s", RedactText(err.Error())) if err := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); err != nil { @@ -1079,13 +1144,29 @@ func (worker *Worker) runJobLoop(ctx context.Context, interval time.Duration) er return ErrSelfUpdateRestartRequested } if !handled && time.Since(claimStartedAt) < interval { - if err := waitWorkerLoop(ctx, interval-time.Since(claimStartedAt)); err != nil { + if err := waitWorkerJobWake(ctx, interval-time.Since(claimStartedAt), wake); err != nil { return err } } } } +func waitWorkerJobWake(ctx context.Context, delay time.Duration, wake <-chan struct{}) error { + if delay <= 0 { + return nil + } + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-wake: + return nil + case <-timer.C: + return nil + } +} + func waitWorkerLoop(ctx context.Context, delay time.Duration) error { if delay <= 0 { return nil diff --git a/runtime/worker_test.go b/runtime/worker_test.go index ee5a7a5..9e30e05 100644 --- a/runtime/worker_test.go +++ b/runtime/worker_test.go @@ -905,6 +905,16 @@ func (client *fakeWorkerClient) Heartbeat(_ context.Context, request protocol.Ru return protocol.RunHeartbeatResponse{Accepted: true, RunEndpointID: request.RunEndpointID, NextHeartbeatSeconds: 15, ServerTime: workerTestTime()}, nil } +func (client *fakeWorkerClient) StreamControlEvents(ctx context.Context, request protocol.RunControlStreamRequest, handle func(protocol.RunControlEvent) error) error { + if handle != nil { + if err := handle(protocol.RunControlEvent{RunEndpointID: request.RunEndpointID, Type: protocol.RunControlEventTypeReady, Sequence: request.LastEventSeq, ServerTime: workerTestTime()}); err != nil { + return err + } + } + <-ctx.Done() + return ctx.Err() +} + func (client *fakeWorkerClient) ReportLifecycle(_ context.Context, request protocol.RunLifecycleReportRequest) (protocol.RunLifecycleReportResponse, error) { client.lifecycleReports = append(client.lifecycleReports, request) if client.lifecycleReportErr != nil {