From 330b1c01303dd979d4082f829670b8651ac0dc6a Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Thu, 3 Sep 2026 16:40:05 +0800 Subject: [PATCH] Keep run logs opaque and streamline transfers --- README.md | 11 +- api/artifact_client_test.go | 23 ++- api/channel_isolation_test.go | 13 +- api/platform_client.go | 66 ++++++- config/config_test.go | 2 +- config/package_config.go | 45 +---- config/package_config_test.go | 34 +++- protocol/artifact.go | 2 +- protocol/artifact.md | 6 +- protocol/job.go | 2 +- protocol/job.md | 2 +- runtime/autonomous_lifecycle.go | 20 +- runtime/distribution_build.go | 280 +++++----------------------- runtime/distribution_build_test.go | 162 ++-------------- runtime/file_artifact_transfer.go | 2 +- runtime/lifecycle.go | 74 +++----- runtime/lifecycle_test.go | 32 +--- runtime/log_sources.go | 10 +- runtime/metrics.go | 4 +- runtime/process_supervisor.go | 38 ++-- runtime/process_window_windows.go | 33 +++- runtime/runtime_profiles.go | 29 +-- runtime/sqlite_schema_probe_test.go | 2 +- runtime/worker.go | 90 ++++----- runtime/workspace_seed.go | 8 +- spool/artifact_queue.go | 98 +++++----- spool/artifact_queue_test.go | 14 ++ 27 files changed, 429 insertions(+), 673 deletions(-) diff --git a/README.md b/README.md index 57e1c7a..45eab60 100644 --- a/README.md +++ b/README.md @@ -60,13 +60,13 @@ go run ./cmd/run Use `RUN_MODE=worker` when you want the executor to register, heartbeat, claim jobs, and execute lifecycle templates. Use `RUN_MODE=smoke` for a one-shot config summary. -Generated run and client-manager packages carry a secret-bearing JSON config created by platform. The config contains: +Generated run packages carry a secret-bearing JSON config created by platform. The config contains: -- component kind: `run` or `client-manager`. -- server instance ID, plugin ID, optional run endpoint ID, optional client-manager profile key. -- target OS/architecture, redacted `secret://runtime-keys/.../current` ref, key generation, and the raw current auth key needed by the remote executable. +- component kind: `run`. +- server instance ID, plugin ID, and optional run endpoint ID. +- target OS/architecture, logical `secret://runtime-keys/.../current` ref, key generation, and the raw current auth key needed by the remote executable. -The raw auth key is valid only while it matches the single current encrypted key stored in platform for that server/component. Resetting the run key or a client-manager key increments generation and makes older packages fail control hello authentication until the operator regenerates and redeploys the affected package. Local diagnostics and smoke summaries use fingerprints and secret refs, not raw keys. +The raw auth key is valid only while it matches the single current encrypted run key stored in platform for that server. Resetting the run key increments generation and makes older packages fail control hello authentication until the operator regenerates and redeploys the affected package. Local diagnostics and smoke summaries use fingerprints and secret refs, not raw keys. 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`. @@ -83,7 +83,6 @@ Run resolves plugin-declared runtime profiles using server runtime bindings supp - `local-process`: run starts/stops the third-party server through scoped lifecycle action refs and tails stdout/stderr. - `hosted-ftp-rcon`: run exposes only declared FTP/log/RCON adapters for hosted servers that cannot be started locally. - `ftp-only`: run exposes declared FTP and log transfer surfaces without lifecycle or RCON control. -- `custom-client`: run coordinates with a plugin-declared companion client manager using a separate component key and profile ref. Profile resolution returns logical capabilities, transport keys, declared log sources, discovery probes, and missing binding keys. It must not return raw host paths, FTP credentials, SQL DSNs, RCON passwords, direct sockets, or component auth keys. diff --git a/api/artifact_client_test.go b/api/artifact_client_test.go index a251ed7..79d91f2 100644 --- a/api/artifact_client_test.go +++ b/api/artifact_client_test.go @@ -3,6 +3,7 @@ package api import ( "context" "encoding/json" + "io" "net/http" "net/http/httptest" "testing" @@ -10,7 +11,7 @@ import ( "browser.local/run/protocol" ) -func TestPlatformClientArtifactMethodsPostJSONAndDecodeResponses(t *testing.T) { +func TestPlatformClientArtifactMethodsUploadRawChunksAndDecodeResponses(t *testing.T) { seen := map[string]bool{} server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { seen[r.URL.Path] = true @@ -23,10 +24,16 @@ func TestPlatformClientArtifactMethodsPostJSONAndDecodeResponses(t *testing.T) { } writeTestJSON(t, w, validArtifactOpenResponse()) case "/api/v1/run/artifacts/chunks": - var request protocol.ArtifactChunkUploadRequest - decodeTestRequest(t, r, &request) - if request.TransferID != "transfer-1" || request.ChunkIndex != 0 || string(request.Payload) != "payload" { - t.Fatalf("unexpected artifact chunk request: %+v", request) + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatalf("read artifact chunk body: %v", err) + } + verifyRunRequestSignature(t, r, body, "run-local", "session-token") + if r.Header.Get("Content-Type") != "application/octet-stream" || r.Header.Get("X-Run-Session-Token") != "session-token" { + t.Fatalf("unexpected artifact chunk content headers: %+v", r.Header) + } + if r.Header.Get("X-Artifact-Transfer-Id") != "transfer-1" || r.Header.Get("X-Artifact-Chunk-Index") != "0" || string(body) != "payload" { + t.Fatalf("unexpected artifact chunk request headers=%+v body=%q", r.Header, string(body)) } writeTestJSON(t, w, protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: "transfer-1", ArtifactID: "artifact-1", ChunkIndex: 0, ReceivedChunkIndexes: []int{0}, NextMissingChunkIndex: 1, ServerTime: fixedClientTestTime()}) case "/api/v1/run/artifacts/status": @@ -95,13 +102,13 @@ func TestPlatformClientArtifactMethodReturnsErrorForPlatformFailure(t *testing.T } } -func TestArtifactChunkPayloadUsesJSONBase64Encoding(t *testing.T) { +func TestArtifactChunkPayloadIsNotJSONEncoded(t *testing.T) { encoded, err := json.Marshal(validClientArtifactChunk()) if err != nil { t.Fatalf("marshal artifact chunk: %v", err) } - if !json.Valid(encoded) || !containsJSONPayloadField(encoded) { - t.Fatalf("expected JSON encoded payload field, got %s", string(encoded)) + if !json.Valid(encoded) || containsJSONPayloadField(encoded) { + t.Fatalf("artifact chunk payload must stay out of JSON, got %s", string(encoded)) } } diff --git a/api/channel_isolation_test.go b/api/channel_isolation_test.go index 6277c9f..f521e9d 100644 --- a/api/channel_isolation_test.go +++ b/api/channel_isolation_test.go @@ -3,6 +3,7 @@ package api import ( "context" "encoding/json" + "io" "net/http" "net/http/httptest" "strings" @@ -20,14 +21,16 @@ func TestPlatformClientLightweightChannelsCompleteWhileArtifactChunkIsBlocked(t server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/api/v1/run/artifacts/chunks": - var request protocol.ArtifactChunkUploadRequest - decodeTestRequest(t, r, &request) - if request.ChunkIndex != 0 || len(request.Payload) == 0 { - t.Fatalf("unexpected artifact payload: %+v", request) + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatalf("read artifact payload: %v", err) + } + if r.Header.Get("X-Artifact-Chunk-Index") != "0" || len(body) == 0 || r.Header.Get("Content-Type") != "application/octet-stream" { + t.Fatalf("unexpected artifact payload headers=%+v body=%q", r.Header, string(body)) } close(artifactStarted) <-releaseArtifact - writeTestJSON(t, w, protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: request.TransferID, ArtifactID: request.ArtifactID, ChunkIndex: request.ChunkIndex, ReceivedChunkIndexes: []int{0}, NextMissingChunkIndex: 1, ServerTime: fixedClientTestTime()}) + writeTestJSON(t, w, protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: r.Header.Get("X-Artifact-Transfer-Id"), ArtifactID: r.Header.Get("X-Artifact-Id"), ChunkIndex: 0, ReceivedChunkIndexes: []int{0}, NextMissingChunkIndex: 1, ServerTime: fixedClientTestTime()}) close(artifactDone) case "/api/v1/run/control/heartbeat": var request protocol.RunHeartbeatRequest diff --git a/api/platform_client.go b/api/platform_client.go index 4585175..e1b9b09 100644 --- a/api/platform_client.go +++ b/api/platform_client.go @@ -31,6 +31,16 @@ type PlatformClient struct { serverClockOffsetActive *atomic.Bool } +const ( + runSessionTokenHeader = "X-Run-Session-Token" + artifactTransferIDHeader = "X-Artifact-Transfer-Id" + artifactIDHeader = "X-Artifact-Id" + artifactChunkIndexHeader = "X-Artifact-Chunk-Index" + artifactChunkOffsetHeader = "X-Artifact-Offset" + artifactChunkSizeHeader = "X-Artifact-Size" + artifactChunkHashHeader = "X-Artifact-Checksum" +) + type PlatformRequestError struct { Status int Path string @@ -326,7 +336,7 @@ func (c PlatformClient) OpenArtifactTransfer(ctx context.Context, request protoc } func (c PlatformClient) UploadArtifactChunk(ctx context.Context, request protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error) { - return postPlatformJSON[protocol.ArtifactChunkUploadRequest, protocol.ArtifactChunkUploadResponse](ctx, c, "/api/v1/run/artifacts/chunks", request) + return postPlatformArtifactChunk(ctx, c, "/api/v1/run/artifacts/chunks", request) } func (c PlatformClient) QueryArtifactTransferStatus(ctx context.Context, request protocol.ArtifactTransferStatusRequest) (protocol.ArtifactTransferStatusResponse, error) { @@ -337,6 +347,56 @@ func (c PlatformClient) CompleteArtifactTransfer(ctx context.Context, request pr return postPlatformJSON[protocol.ArtifactTransferCompleteRequest, protocol.ArtifactTransferCompleteResponse](ctx, c, "/api/v1/run/artifacts/complete", request) } +func postPlatformArtifactChunk(ctx context.Context, client PlatformClient, path string, request protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error) { + var response protocol.ArtifactChunkUploadResponse + startedAt := time.Now() + log.Printf("RUN platform request status=starting method=POST base=%s path=%s", diagnosticLogValue(client.baseURL), path) + + body := request.Payload + httpRequest, err := http.NewRequestWithContext(ctx, http.MethodPost, client.baseURL+path, bytes.NewReader(body)) + if err != nil { + log.Printf("RUN platform request status=build_error method=POST base=%s path=%s durationMs=%d error=%s", diagnosticLogValue(client.baseURL), path, time.Since(startedAt).Milliseconds(), err) + return response, fmt.Errorf("build platform request: %w", err) + } + httpRequest.Header.Set("Content-Type", "application/octet-stream") + httpRequest.Header.Set("Accept", "application/json") + httpRequest.Header.Set(runSessionTokenHeader, request.SessionToken) + httpRequest.Header.Set(artifactTransferIDHeader, request.TransferID) + httpRequest.Header.Set(artifactIDHeader, request.ArtifactID) + httpRequest.Header.Set(artifactChunkIndexHeader, strconv.Itoa(request.ChunkIndex)) + httpRequest.Header.Set(artifactChunkOffsetHeader, strconv.FormatInt(request.Offset, 10)) + httpRequest.Header.Set(artifactChunkSizeHeader, strconv.Itoa(request.SizeBytes)) + httpRequest.Header.Set(artifactChunkHashHeader, request.Checksum) + signatureSummary, err := signRunRequestWithEnvelope(httpRequest, body, runRequestEnvelope{RunEndpointID: request.RunEndpointID, SessionToken: request.SessionToken}, client.signatureTime()) + if err != nil { + log.Printf("RUN platform request status=sign_error method=POST base=%s path=%s durationMs=%d error=%s", diagnosticLogValue(client.baseURL), path, time.Since(startedAt).Milliseconds(), err) + return response, err + } + log.Printf("RUN platform request status=signed method=POST base=%s path=%s endpoint=%s timestamp=%s nonce=%s bodyHash=%s signature=%s", diagnosticLogValue(client.baseURL), path, diagnosticLogValue(signatureSummary.RunEndpointID), signatureSummary.Timestamp, shortDiagnosticValue(signatureSummary.Nonce), shortDiagnosticValue(signatureSummary.BodyHash), shortDiagnosticValue(signatureSummary.Signature)) + + httpResponse, err := client.httpClient.Do(httpRequest) + if err != nil { + log.Printf("RUN platform request status=send_error method=POST base=%s path=%s durationMs=%d error=%s", diagnosticLogValue(client.baseURL), path, time.Since(startedAt).Milliseconds(), err) + return response, fmt.Errorf("send platform request: %w", err) + } + defer httpResponse.Body.Close() + log.Printf("RUN platform request status=response method=POST base=%s path=%s httpStatus=%d durationMs=%d", diagnosticLogValue(client.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 response, PlatformRequestError{Status: httpResponse.StatusCode, Path: path, Code: failure.Code, Details: failure.Details} + } + if err := json.NewDecoder(httpResponse.Body).Decode(&response); err != nil { + return response, fmt.Errorf("decode platform response: %w", err) + } + client.observeResponseServerTime(response) + return response, nil +} + func postPlatformJSON[Request any, Response any](ctx context.Context, client PlatformClient, path string, request Request) (Response, error) { var response Response startedAt := time.Now() @@ -405,6 +465,10 @@ func signRunRequest(request *http.Request, body []byte, stamp time.Time) (runReq if err := json.Unmarshal(body, &envelope); err != nil { return runRequestSignatureSummary{}, fmt.Errorf("decode Run signing envelope: %w", err) } + return signRunRequestWithEnvelope(request, body, envelope, stamp) +} + +func signRunRequestWithEnvelope(request *http.Request, body []byte, envelope runRequestEnvelope, stamp time.Time) (runRequestSignatureSummary, error) { if strings.TrimSpace(envelope.RunEndpointID) == "" || strings.TrimSpace(envelope.SessionToken) == "" { return runRequestSignatureSummary{}, fmt.Errorf("Run signing envelope requires endpoint and session token") } diff --git a/config/config_test.go b/config/config_test.go index 75b85cd..668b00c 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -89,7 +89,7 @@ func TestLoadUsesPackagedIdentityOverEnvironment(t *testing.T) { t.Setenv("RUN_REGISTRATION_TOKEN", "stale-token") t.Setenv("RUN_SERVER_INSTANCE_ID", "stale-server") t.Setenv("RUN_PLUGIN_ID", "stale-plugin") - t.Setenv("RUN_COMPONENT_KIND", "client-manager") + t.Setenv("RUN_COMPONENT_KIND", "stale-kind") t.Setenv("RUN_COMPONENT_KEY", "stale-profile") t.Setenv("RUN_KEY_GENERATION", "7") t.Setenv("RUN_VERSION", "stale-version") diff --git a/config/package_config.go b/config/package_config.go index 3a897c8..0209656 100644 --- a/config/package_config.go +++ b/config/package_config.go @@ -11,8 +11,7 @@ import ( ) const ( - PackageComponentRun = "run" - PackageComponentClientManager = "client-manager" + PackageComponentRun = "run" PackageConfigEnv = "RUN_PACKAGE_CONFIG" ) @@ -94,7 +93,7 @@ func LoadPackageConfigBesideExecutable() (PackageConfig, bool, error) { func ValidatePackageConfig(cfg PackageConfig) error { var violations []string - if cfg.Kind != PackageComponentRun && cfg.Kind != PackageComponentClientManager { + if cfg.Kind != PackageComponentRun { violations = append(violations, "kind is invalid") } if !safeIdentifier(cfg.ServerInstanceID) { @@ -109,13 +108,10 @@ func ValidatePackageConfig(cfg PackageConfig) error { if cfg.ProfileKey != "" && !safeLogicalKey(cfg.ProfileKey) { violations = append(violations, "profileKey is invalid") } - if cfg.Kind == PackageComponentClientManager && cfg.ProfileKey == "" { - violations = append(violations, "profileKey is required for client-manager packages") - } if !safeRuntimeTarget(cfg.TargetOS, cfg.TargetArch) { violations = append(violations, "target platform is invalid") } - if !strings.HasPrefix(cfg.SecretRef, "secret://runtime-keys/") || containsUnsafeDiagnosticText(cfg.SecretRef) { + if !strings.HasPrefix(cfg.SecretRef, "secret://runtime-keys/") || strings.ContainsAny(cfg.SecretRef, " \t\r\n") { violations = append(violations, "secretRef is invalid") } if cfg.KeyGeneration <= 0 { @@ -124,9 +120,6 @@ func ValidatePackageConfig(cfg PackageConfig) error { if strings.TrimSpace(cfg.AuthKey) == "" { violations = append(violations, "authKey is required") } - if containsUnsafeDiagnosticText(cfg.AuthKey) { - violations = append(violations, "authKey contains unsafe content") - } if len(violations) > 0 { return fmt.Errorf("invalid run package config: %s", strings.Join(violations, "; ")) } @@ -165,7 +158,7 @@ func (cfg PackageConfig) Identity() PackageIdentity { } } -func (cfg PackageConfig) RedactedDiagnostics() map[string]string { +func (cfg PackageConfig) IdentityDiagnostics() map[string]string { identity := cfg.Identity() return map[string]string{ "kind": identity.Kind, @@ -188,7 +181,7 @@ func AuthenticatePackageGeneration(pkg PackageConfig, auth ComponentAuthResult) return fmt.Errorf("component authentication scope does not match package") } if !auth.Allowed { - return fmt.Errorf("component authentication rejected: %s", redactedReason(auth.Reason)) + return fmt.Errorf("component authentication rejected: %s", auth.Reason) } if auth.KeyGeneration != pkg.KeyGeneration { return fmt.Errorf("component key generation is no longer current") @@ -196,11 +189,8 @@ func AuthenticatePackageGeneration(pkg PackageConfig, auth ComponentAuthResult) return nil } -func packageComponentKey(pkg PackageConfig) string { - if pkg.Kind == PackageComponentRun { - return "" - } - return strings.TrimSpace(pkg.ProfileKey) +func packageComponentKey(PackageConfig) string { + return "" } func fingerprint(value string) string { @@ -210,7 +200,7 @@ func fingerprint(value string) string { func safeIdentifier(value string) bool { value = strings.TrimSpace(value) - if value == "" || len(value) > 120 || containsUnsafeDiagnosticText(value) { + if value == "" || len(value) > 120 { return false } for _, char := range value { @@ -228,7 +218,7 @@ func safePluginID(value string) bool { func safeLogicalKey(value string) bool { value = strings.TrimSpace(value) - if value == "" || len(value) > 120 || strings.HasPrefix(value, "/") || strings.Contains(value, "..") || strings.Contains(value, `\`) || containsUnsafeDiagnosticText(value) { + if value == "" || len(value) > 120 || strings.HasPrefix(value, "/") || strings.Contains(value, "..") || strings.Contains(value, `\`) { return false } for _, char := range value { @@ -253,20 +243,3 @@ func safeRuntimeTarget(osName string, arch string) bool { return false } } - -func containsUnsafeDiagnosticText(value string) bool { - normalized := strings.ToLower(value) - for _, marker := range []string{"/users/", "/.ssh/", "password=", "apikey", "api_key", "bearer ", "sk-", "unix://", "tcp://", "mysql://", "sqlite://"} { - if strings.Contains(normalized, marker) { - return true - } - } - return false -} - -func redactedReason(value string) string { - if containsUnsafeDiagnosticText(value) { - return "[redacted]" - } - return value -} diff --git a/config/package_config_test.go b/config/package_config_test.go index 8e1edab..69e0067 100644 --- a/config/package_config_test.go +++ b/config/package_config_test.go @@ -30,14 +30,14 @@ func TestLoadPackageConfigAppliesServerScopedIdentity(t *testing.T) { if cfg.RegistrationToken != "opaque-runtime-key" || cfg.RunEndpointID != "run-server-1" || cfg.ServerInstanceID != "server-1" || cfg.ComponentKey != "" || cfg.KeyGeneration != 3 { t.Fatalf("expected package identity to be applied, got %+v", cfg) } - diagnostics := pkg.RedactedDiagnostics() + diagnostics := pkg.IdentityDiagnostics() for _, value := range diagnostics { if strings.Contains(value, "opaque-runtime-key") || strings.Contains(value, "/Users/") || strings.Contains(value, "password=") { t.Fatalf("diagnostics exposed sensitive value: %+v", diagnostics) } } if diagnostics["keyFingerprint"] == "" || diagnostics["secretRef"] != "secret://runtime-keys/server-1/run/current" { - t.Fatalf("expected redacted key fingerprint and secret ref, got %+v", diagnostics) + t.Fatalf("expected key fingerprint and logical secret ref, got %+v", diagnostics) } } @@ -56,8 +56,7 @@ func TestLoadPackageConfigRejectsUnsafeOrIncompletePackages(t *testing.T) { "old zero generation": func(cfg PackageConfig) PackageConfig { cfg.KeyGeneration = 0; return cfg }, "raw path": func(cfg PackageConfig) PackageConfig { cfg.ServerInstanceID = "/Users/tasia/server"; return cfg }, "socket": func(cfg PackageConfig) PackageConfig { cfg.SecretRef = "unix:///tmp/run.sock"; return cfg }, - "secret auth": func(cfg PackageConfig) PackageConfig { cfg.AuthKey = "password=raw"; return cfg }, - "client missing key": func(cfg PackageConfig) PackageConfig { cfg.Kind = PackageComponentClientManager; return cfg }, + "legacy component": func(cfg PackageConfig) PackageConfig { cfg.Kind = "client-manager"; return cfg }, } for name, mutate := range cases { if err := ValidatePackageConfig(mutate(valid)); err == nil { @@ -66,6 +65,22 @@ func TestLoadPackageConfigRejectsUnsafeOrIncompletePackages(t *testing.T) { } } +func TestPackageConfigTreatsAuthKeyAsOpaqueValue(t *testing.T) { + cfg := PackageConfig{ + Kind: PackageComponentRun, + ServerInstanceID: "server-1", + PluginID: "game.scum", + TargetOS: "windows", + TargetArch: "amd64", + SecretRef: "secret://runtime-keys/server-1/run/current", + KeyGeneration: 1, + AuthKey: "password=raw sk-test-value bearer literal", + } + if err := ValidatePackageConfig(cfg); err != nil { + t.Fatalf("authKey must be treated as opaque package data, got %v", err) + } +} + func TestAuthenticatePackageGenerationRejectsStalePackages(t *testing.T) { pkg := PackageConfig{ Kind: PackageComponentRun, @@ -102,15 +117,16 @@ func TestAuthenticatePackageGenerationRejectsStalePackages(t *testing.T) { func TestLoadPackageConfigFromEnv(t *testing.T) { path := writePackageConfig(t, PackageConfig{ - Kind: PackageComponentClientManager, + Kind: PackageComponentRun, ServerInstanceID: "server-1", PluginID: "game.scum", - ProfileKey: "scum-client-manager", + RunEndpointID: "run-server-1", + ProfileKey: "run-local", TargetOS: "windows", TargetArch: "amd64", - SecretRef: "secret://runtime-keys/server-1/client-manager/scum-client-manager/current", + SecretRef: "secret://runtime-keys/server-1/run/current", KeyGeneration: 4, - AuthKey: "opaque-client-key", + AuthKey: "opaque-run-key", }) t.Setenv(PackageConfigEnv, path) @@ -118,7 +134,7 @@ func TestLoadPackageConfigFromEnv(t *testing.T) { if err != nil || !ok { t.Fatalf("expected env package config, ok=%v err=%v", ok, err) } - if cfg.Kind != PackageComponentClientManager || cfg.ProfileKey != "scum-client-manager" { + if cfg.Kind != PackageComponentRun || cfg.ProfileKey != "run-local" { t.Fatalf("unexpected package config: %+v", cfg) } diff --git a/protocol/artifact.go b/protocol/artifact.go index 3474dae..b9e3220 100644 --- a/protocol/artifact.go +++ b/protocol/artifact.go @@ -49,7 +49,7 @@ type ArtifactChunkUploadRequest struct { Offset int64 `json:"offset"` SizeBytes int `json:"sizeBytes"` Checksum string `json:"checksum"` - Payload []byte `json:"payload"` + Payload []byte `json:"-"` } type ArtifactChunkUploadResponse struct { diff --git a/protocol/artifact.md b/protocol/artifact.md index fcef861..0e13e26 100644 --- a/protocol/artifact.md +++ b/protocol/artifact.md @@ -5,7 +5,7 @@ Artifacts move files and large payloads between platform and run without blockin ## Implemented Routes - `POST /api/v1/run/artifacts/open`: opens a run-to-platform upload transfer and returns resume state. -- `POST /api/v1/run/artifacts/chunks`: uploads one bounded chunk with byte range and checksum metadata. +- `POST /api/v1/run/artifacts/chunks`: uploads one bounded octet-stream chunk with byte range and checksum metadata in headers. - `POST /api/v1/run/artifacts/status`: queries received chunks and the next missing chunk index. - `POST /api/v1/run/artifacts/complete`: verifies all chunks and final checksum before marking the artifact available. @@ -18,7 +18,7 @@ Browser-facing artifact downloads are implemented through platform-owned routes - `ArtifactTransferOpenRequest`: run ID, session token, artifact ID, upload direction, owner scope, size, chunk size, checksum, and idempotency key. - `ArtifactTransferOpenResponse`: transfer ID, artifact metadata, total chunks, received chunk indexes, next missing chunk index, duplicate flag, and server time. -- `ArtifactChunkUploadRequest`: transfer ID, artifact ID, chunk index, byte offset, size, checksum, and JSON byte payload. +- `ArtifactChunkUploadRequest`: transfer ID, artifact ID, chunk index, byte offset, size, checksum, and an in-memory octet-stream payload buffer that is not JSON encoded. - `ArtifactChunkUploadResponse`: accepted chunk index, received chunk indexes, next missing chunk index, duplicate flag, and server time. - `ArtifactTransferStatusRequest`: run ID, session token, transfer ID, and artifact ID. - `ArtifactTransferStatusResponse`: transfer direction, total chunks, received chunk indexes, next missing chunk index, completion flag, and server time. @@ -27,7 +27,7 @@ Browser-facing artifact downloads are implemented through platform-owned routes ## Local Queue -Run stores unacknowledged `ArtifactChunkUploadRequest` payloads in the local artifact queue. A queued chunk may be removed only after the platform acknowledges the same transfer ID, artifact ID, and chunk index. The queue must not store or expose raw host paths. +Run stores unacknowledged artifact chunk metadata as JSON and raw chunk bytes as sidecar files in the local artifact queue. A queued chunk may be removed only after the platform acknowledges the same transfer ID, artifact ID, and chunk index. The queue must not store or expose raw host paths. ## Rules diff --git a/protocol/job.go b/protocol/job.go index defb450..373f29c 100644 --- a/protocol/job.go +++ b/protocol/job.go @@ -135,7 +135,7 @@ type SQLiteSchemaProbeSafeError struct { Retryable bool `json:"retryable"` } -// SQLiteSchemaProbeResult is a terminal, redacted envelope. Names and sample +// SQLiteSchemaProbeResult is a terminal, bounded envelope. Names and sample // values are represented only by salted-looking SHA-256 fingerprints. type SQLiteSchemaProbeResult struct { RequestID string `json:"requestId"` diff --git a/protocol/job.md b/protocol/job.md index a488429..ed6ec89 100644 --- a/protocol/job.md +++ b/protocol/job.md @@ -93,4 +93,4 @@ remain separate channels and must not be multiplexed through job result payloads. Artifact transfer carries chunk payloads only through `/api/v1/run/artifacts/*` routes. -Production code signing/KMS, rollout rings/fleet orchestration, client-manager lifecycle, plugin lifecycle, production scaling/alerts, and real AI-provider integration are explicitly outside this contract. +Production code signing/KMS, rollout rings/fleet orchestration, plugin lifecycle, production scaling/alerts, and real AI-provider integration are explicitly outside this contract. diff --git a/runtime/autonomous_lifecycle.go b/runtime/autonomous_lifecycle.go index 8a1f933..953b4d5 100644 --- a/runtime/autonomous_lifecycle.go +++ b/runtime/autonomous_lifecycle.go @@ -27,7 +27,7 @@ func LoadAutonomousLifecyclePlan(cfg config.Config) (*protocol.RunAutonomousLife } scope, err := seededWorkspaceScope(cfg) if err != nil { - log.Printf("RUN phase=autonomous_lifecycle status=scope_failed server=%s componentKey=%s error=%s", safeOptional(cfg.ServerInstanceID), safeOptional(cfg.ComponentKey), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle status=scope_failed server=%s componentKey=%s error=%s", safeOptional(cfg.ServerInstanceID), safeOptional(cfg.ComponentKey), err.Error()) return nil, "", false, err } path, err := NewWorkspaceResolver(cfg.WorkspaceRoot).ExistingTarget(scope, autonomousLifecyclePlanKey) @@ -36,12 +36,12 @@ func LoadAutonomousLifecyclePlan(cfg config.Config) (*protocol.RunAutonomousLife log.Printf("RUN phase=autonomous_lifecycle status=skipped reason=plan_missing scope=%s", safeOptional(scope)) return nil, scope, false, nil } - log.Printf("RUN phase=autonomous_lifecycle status=load_failed scope=%s error=%s", safeOptional(scope), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle status=load_failed scope=%s error=%s", safeOptional(scope), err.Error()) return nil, scope, false, err } file, err := os.Open(path) if err != nil { - log.Printf("RUN phase=autonomous_lifecycle status=open_failed scope=%s error=%s", safeOptional(scope), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle status=open_failed scope=%s error=%s", safeOptional(scope), err.Error()) return nil, scope, false, err } defer file.Close() @@ -49,11 +49,11 @@ func LoadAutonomousLifecyclePlan(cfg config.Config) (*protocol.RunAutonomousLife decoder := json.NewDecoder(io.LimitReader(file, 64*1024)) decoder.DisallowUnknownFields() if err := decoder.Decode(&plan); err != nil { - log.Printf("RUN phase=autonomous_lifecycle status=decode_failed scope=%s error=%s", safeOptional(scope), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle status=decode_failed scope=%s error=%s", safeOptional(scope), err.Error()) return nil, scope, false, err } if err := protocol.ValidateRunAutonomousLifecyclePlan(plan); err != nil { - log.Printf("RUN phase=autonomous_lifecycle status=invalid scope=%s error=%s", safeOptional(scope), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle status=invalid scope=%s error=%s", safeOptional(scope), err.Error()) return nil, scope, false, err } log.Printf("RUN phase=autonomous_lifecycle status=loaded server=%s plugin=%s endpoint=%s profile=%s bootstrap=%t actions=%d dependencies=%d installs=%d", plan.ServerInstanceID, plan.PluginID, plan.RunEndpointID, safeOptional(plan.ProfileKey), plan.Bootstrap != nil, len(plan.Actions), len(plan.DependencyProbes), len(plan.InstallPlans)) @@ -70,7 +70,7 @@ func (worker *Worker) RunAutonomousLifecycleOnce(ctx context.Context) error { return err } if err := validateAutonomousLifecycleScope(worker.cfg, state, *plan); err != nil { - log.Printf("RUN phase=autonomous_lifecycle status=scope_mismatch error=%s", RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle status=scope_mismatch error=%s", err.Error()) return err } if err := worker.runAutonomousDependencies(ctx, *plan); err != nil { @@ -124,7 +124,7 @@ func (worker *Worker) reportAutonomousLifecycle(ctx context.Context, assignment log.Printf("RUN phase=autonomous_lifecycle.report status=starting server=%s capability=%s state=%s processState=%s", assignment.ServerInstanceID, assignment.Capability, execution.State, safeOptional(execution.ExecutionResult.ProcessState)) response, err := worker.client.ReportLifecycle(ctx, request) if err != nil { - log.Printf("RUN phase=autonomous_lifecycle.report status=failed server=%s error=%s", assignment.ServerInstanceID, RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle.report status=failed server=%s error=%s", assignment.ServerInstanceID, err.Error()) return err } if !response.Accepted || response.RunEndpointID != state.RunEndpointID || response.ServerInstanceID != assignment.ServerInstanceID { @@ -196,10 +196,10 @@ func (worker *Worker) runAutonomousDependencies(ctx context.Context, plan protoc state, evidence, err := worker.executor.runDependencyProbe(ctx, probe, plan.RuntimeBindings) if err != nil { if probe.Required { - log.Printf("RUN phase=autonomous_lifecycle.dependencies status=failed probe=%s error=%s", safeOptional(probe.Key), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle.dependencies status=failed probe=%s error=%s", safeOptional(probe.Key), err.Error()) return err } - log.Printf("RUN phase=autonomous_lifecycle.dependencies status=optional_probe_failed probe=%s error=%s", safeOptional(probe.Key), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle.dependencies status=optional_probe_failed probe=%s error=%s", safeOptional(probe.Key), err.Error()) continue } log.Printf("RUN phase=autonomous_lifecycle.dependencies status=probed probe=%s state=%s evidence=%s required=%t", safeOptional(probe.Key), safeOptional(state), safeOptional(evidence), probe.Required) @@ -215,7 +215,7 @@ func (worker *Worker) runAutonomousDependencies(ctx context.Context, plan protoc input := protocol.DependencyExecutionInputResponse{JobID: assignment.JobID, ServerInstanceID: plan.ServerInstanceID, RunEndpointID: plan.RunEndpointID, PluginID: plan.PluginID, PluginVersion: plan.PluginVersion, ProfileKey: autonomousProfileKey(worker.cfg, plan), TargetOS: plan.TargetOS, TargetArch: plan.TargetArch, PlanDigest: autonomousInstallPlanDigest(installPlan), Probe: probe, Plan: installPlan, Bindings: plan.RuntimeBindings} log.Printf("RUN phase=autonomous_lifecycle.dependencies status=install_start probe=%s plan=%s steps=%d", safeOptional(probe.Key), safeOptional(installPlan.Key), len(installPlan.Steps)) if err := worker.runAutonomousInstallPlan(ctx, assignment, input); err != nil { - log.Printf("RUN phase=autonomous_lifecycle.dependencies status=install_failed probe=%s plan=%s error=%s", safeOptional(probe.Key), safeOptional(installPlan.Key), RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle.dependencies status=install_failed probe=%s plan=%s error=%s", safeOptional(probe.Key), safeOptional(installPlan.Key), err.Error()) return err } log.Printf("RUN phase=autonomous_lifecycle.dependencies status=install_complete probe=%s plan=%s", safeOptional(probe.Key), safeOptional(installPlan.Key)) diff --git a/runtime/distribution_build.go b/runtime/distribution_build.go index aecd039..d982cd7 100644 --- a/runtime/distribution_build.go +++ b/runtime/distribution_build.go @@ -1,9 +1,6 @@ package runtime import ( - "archive/tar" - "archive/zip" - "compress/gzip" "context" "crypto/sha256" "encoding/base64" @@ -19,7 +16,6 @@ import ( "strings" "time" - "browser.local/run/config" "browser.local/run/protocol" ) @@ -108,15 +104,6 @@ func (worker *Worker) executeDistributionBuild(ctx context.Context, assignment p return distributionBuildFailure("env_check_failed", "Go build environment is unavailable") } - configPath := "" - if input.ComponentKind == "client-manager" { - var err error - configPath, err = writeDistributionConfig(sourceRoot, input, worker.clientPlatformURL()) - if err != nil { - return distributionBuildFailure("config_injection_failed", "could not inject scoped component configuration") - } - } - if err := report(45, "deps_download: downloading Go modules"); err != nil { return distributionBuildFailure("progress_report_failed", "could not report dependency download") } @@ -129,14 +116,8 @@ func (worker *Worker) executeDistributionBuild(ctx context.Context, assignment p return distributionBuildFailure("progress_report_failed", "could not report compilation") } binaryPath := filepath.Join(workspace, input.OutputFilename) - entry := "." - if input.ComponentKind == "run" { - entry = "./cmd/run" - } - ldflags := "-s -w" - if input.ComponentKind == "run" { - ldflags = buildRunLDFlags(input, distributionBuildPlatformURL(worker, input)) - } + entry := "./cmd/run" + ldflags := buildRunLDFlags(input, distributionBuildPlatformURL(worker, input)) if err := fixedCommand(ctx, sourceRoot, buildEnv, "go", "build", "-trimpath", "-ldflags", ldflags, "-o", binaryPath, entry); err != nil { return distributionBuildFailure("build_compile_failed", "Go compilation failed") } @@ -144,33 +125,7 @@ func (worker *Worker) executeDistributionBuild(ctx context.Context, assignment p if err := report(82, "package_finalize: creating distribution archive"); err != nil { return distributionBuildFailure("progress_report_failed", "could not report packaging") } - if input.ComponentKind == "run" { - payload, err := os.ReadFile(binaryPath) - if err != nil { - return distributionBuildFailure("package_finalize_failed", "run executable could not be read") - } - if err := worker.uploadDistributionArtifact(ctx, assignment, input.ArtifactID, payload); err != nil { - return distributionBuildFailure("artifact_upload_failed", "distribution artifact upload failed") - } - if err := report(96, "package_finalize: artifact upload completed"); err != nil { - return distributionBuildFailure("progress_report_failed", "could not report artifact upload") - } - return LifecycleExecutionResult{ - State: lifecycleResultStateSucceeded, - Progress: protocol.RunJobProgressReport{Percent: 100, Message: "package_finalize: build artifact available"}, - ResultRef: "artifact://" + input.ArtifactID, - Message: "distribution build completed", - } - } - archivePath := filepath.Join(workspace, archiveFilename(input)) - if err := createDistributionArchive(archivePath, input.PackageFormat, binaryPath, configPath); err != nil { - return distributionBuildFailure("package_finalize_failed", "distribution archive creation failed") - } - payload, err := os.ReadFile(archivePath) - if err != nil { - return distributionBuildFailure("package_finalize_failed", "distribution archive could not be read") - } - if err := worker.uploadDistributionArtifact(ctx, assignment, input.ArtifactID, payload); err != nil { + if err := worker.uploadDistributionArtifact(ctx, assignment, input.ArtifactID, binaryPath); err != nil { return distributionBuildFailure("artifact_upload_failed", "distribution artifact upload failed") } if err := report(96, "package_finalize: artifact upload completed"); err != nil { @@ -184,45 +139,26 @@ func (worker *Worker) executeDistributionBuild(ctx context.Context, assignment p } } -func (worker *Worker) prepareDistributionSource(ctx context.Context, workspace string, input protocol.DistributionBuildInputResponse) (string, error) { - if input.ComponentKind == "run" { - root, err := filepath.Abs(worker.cfg.BuildSourceRoot) - if err != nil { - return "", err - } - root, err = filepath.EvalSymlinks(root) - if err != nil { - return "", err - } - if _, err := os.Stat(filepath.Join(root, "go.mod")); err != nil { - return "", err - } - isolatedSource := filepath.Join(workspace, "source") - if err := copyDistributionSource(root, isolatedSource, workspace); err != nil { - return "", err - } - if err := writeRunWorkspaceSeedConfig(isolatedSource, input.WorkspaceSeed); err != nil { - return "", err - } - return isolatedSource, nil - } - checkout := filepath.Join(workspace, "source") - if err := os.MkdirAll(checkout, 0o700); err != nil { +func (worker *Worker) prepareDistributionSource(_ context.Context, workspace string, input protocol.DistributionBuildInputResponse) (string, error) { + root, err := filepath.Abs(worker.cfg.BuildSourceRoot) + if err != nil { return "", err } - if err := fixedCommand(ctx, checkout, nil, "git", "init", "--quiet"); err != nil { + root, err = filepath.EvalSymlinks(root) + if err != nil { return "", err } - if err := fixedCommand(ctx, checkout, nil, "git", "remote", "add", "origin", input.RepositoryURL); err != nil { + if _, err := os.Stat(filepath.Join(root, "go.mod")); err != nil { return "", err } - if err := fixedCommand(ctx, checkout, nil, "git", "fetch", "--quiet", "--depth", "1", "origin", input.SourceRevision); err != nil { + isolatedSource := filepath.Join(workspace, "source") + if err := copyDistributionSource(root, isolatedSource, workspace); err != nil { return "", err } - if err := fixedCommand(ctx, checkout, nil, "git", "checkout", "--quiet", "--detach", "FETCH_HEAD"); err != nil { + if err := writeRunWorkspaceSeedConfig(isolatedSource, input.WorkspaceSeed); err != nil { return "", err } - return checkout, nil + return isolatedSource, nil } func writeRunWorkspaceSeedConfig(sourceRoot string, encodedSeed string) error { @@ -302,18 +238,6 @@ func copyDistributionSource(sourceRoot string, destinationRoot string, workspace }) } -func writeDistributionConfig(sourceRoot string, input protocol.DistributionBuildInputResponse, platformURL string) (string, error) { - if input.ComponentKind == "client-manager" { - content := fmt.Sprintf("server_url: %q\nserver_instance_id: %q\nscum_client_credential: %q\nscum_client_name: %q\nscum_client_version: %q\nscum_client_machine_label: %q\nftp_provider: 3\n", - platformURL, input.ServerInstanceID, input.AuthKey, input.ProfileKey, "platform-build", "managed-client") - if err := os.WriteFile(filepath.Join(sourceRoot, "config.yaml"), []byte(content), 0o600); err != nil { - return "", err - } - return filepath.Join(sourceRoot, "config.yaml"), nil - } - return "", fmt.Errorf("run distributions do not use sidecar package config") -} - func distributionBuildPlatformURL(worker *Worker, input protocol.DistributionBuildInputResponse) string { if value := strings.TrimSpace(input.PlatformURL); value != "" { return value @@ -342,15 +266,15 @@ func buildRunLDFlags(input protocol.DistributionBuildInputResponse, platformURL return strings.Join(flags, " ") } -func runBuildComponentKey(input protocol.DistributionBuildInputResponse) string { - if input.ComponentKind == config.PackageComponentRun { - return "" - } - return strings.TrimSpace(input.ProfileKey) +func runBuildComponentKey(protocol.DistributionBuildInputResponse) string { + return "" } -func (worker *Worker) uploadDistributionArtifact(ctx context.Context, assignment protocol.RunJobAssignment, artifactID string, payload []byte) error { - checksum := bytesChecksum(payload) +func (worker *Worker) uploadDistributionArtifact(ctx context.Context, assignment protocol.RunJobAssignment, artifactID string, artifactPath string) error { + checksum, sizeBytes, err := checksumFile(artifactPath) + if err != nil { + return err + } state, err := worker.registeredState() if err != nil { return err @@ -362,7 +286,7 @@ func (worker *Worker) uploadDistributionArtifact(ctx context.Context, assignment Direction: "upload", OwnerKind: "job", OwnerID: assignment.JobID, - SizeBytes: int64(len(payload)), + SizeBytes: sizeBytes, ChunkSizeBytes: distributionArtifactChunkSize, Checksum: checksum, IdempotencyKey: "distribution-build:" + assignment.JobID, @@ -374,7 +298,13 @@ func (worker *Worker) uploadDistributionArtifact(ctx context.Context, assignment for _, index := range opened.ReceivedChunkIndexes { received[index] = true } - for index, offset := 0, 0; offset < len(payload); index, offset = index+1, offset+distributionArtifactChunkSize { + file, err := os.Open(artifactPath) + if err != nil { + return err + } + defer file.Close() + buffer := make([]byte, distributionArtifactChunkSize) + for index, offset := 0, int64(0); offset < sizeBytes; index, offset = index+1, offset+int64(distributionArtifactChunkSize) { if received[index] { continue } @@ -382,18 +312,25 @@ func (worker *Worker) uploadDistributionArtifact(ctx context.Context, assignment if err != nil { return err } - end := offset + distributionArtifactChunkSize - if end > len(payload) { - end = len(payload) + length := distributionArtifactChunkSize + if remaining := sizeBytes - offset; remaining < int64(length) { + length = int(remaining) } - chunk := payload[offset:end] + read, err := file.ReadAt(buffer[:length], offset) + if err != nil && !(err == io.EOF && read == length) { + return err + } + if read != length { + return fmt.Errorf("distribution artifact chunk is shorter than expected") + } + chunk := buffer[:length] if _, err := worker.client.UploadArtifactChunk(ctx, protocol.ArtifactChunkUploadRequest{ RunEndpointID: state.RunEndpointID, SessionToken: state.SessionToken, TransferID: opened.TransferID, ArtifactID: artifactID, ChunkIndex: index, - Offset: int64(offset), + Offset: offset, SizeBytes: len(chunk), Checksum: bytesChecksum(chunk), Payload: chunk, @@ -411,7 +348,7 @@ func (worker *Worker) uploadDistributionArtifact(ctx context.Context, assignment TransferID: opened.TransferID, ArtifactID: artifactID, Checksum: checksum, - SizeBytes: int64(len(payload)), + SizeBytes: sizeBytes, }) if err != nil { return err @@ -426,24 +363,15 @@ func validateDistributionBuildInput(assignment protocol.RunJobAssignment, input if input.JobID != assignment.JobID || input.ServerInstanceID != assignment.ServerInstanceID { return fmt.Errorf("build input scope does not match job") } - if input.ComponentKind != "run" && input.ComponentKind != "client-manager" { + if input.ComponentKind != "run" { return fmt.Errorf("build component kind is unsupported") } if !protocol.ValidLogicalFileKey(input.RunEndpointID) { return fmt.Errorf("generated Run endpoint identity is unsafe") } - if input.ComponentKind == "client-manager" && input.RunEndpointID != assignment.RunEndpointID { - return fmt.Errorf("client-manager build target does not match job") - } if strings.TrimSpace(input.PluginID) == "" { return fmt.Errorf("build plugin id is required") } - if input.ComponentKind == "client-manager" && !approvedHTTPSGitRepository(input.RepositoryURL) { - return fmt.Errorf("client-manager repository is not approved") - } - if input.ComponentKind == "client-manager" && strings.TrimSpace(input.SourceRevision) == "" { - return fmt.Errorf("client-manager source revision is required") - } if input.TargetOS != "windows" && input.TargetOS != "linux" && input.TargetOS != "darwin" { return fmt.Errorf("target OS is unsupported") } @@ -464,20 +392,12 @@ func validateDistributionBuildInput(assignment protocol.RunJobAssignment, input return fmt.Errorf("run workspace seed is invalid") } } - if input.ComponentKind == "client-manager" && input.PackageFormat != "zip" && input.PackageFormat != "tar.gz" { - return fmt.Errorf("package format is unsupported") - } if strings.TrimSpace(input.ArtifactID) == "" || strings.TrimSpace(input.OutputFilename) == "" || strings.TrimSpace(input.AuthKey) == "" { return fmt.Errorf("build input is incomplete") } return nil } -func approvedHTTPSGitRepository(value string) bool { - parsed, err := url.ParseRequestURI(strings.TrimSpace(value)) - return err == nil && parsed.Scheme == "https" && parsed.Host != "" && parsed.User == nil && parsed.RawQuery == "" && parsed.Fragment == "" && strings.HasSuffix(parsed.Path, ".git") -} - func validDistributionPlatformURL(value string) bool { parsed, err := url.ParseRequestURI(strings.TrimSpace(value)) return err == nil && (parsed.Scheme == "https" || parsed.Scheme == "http") && parsed.Host != "" && parsed.User == nil && parsed.RawQuery == "" && parsed.Fragment == "" @@ -485,7 +405,7 @@ func validDistributionPlatformURL(value string) bool { func fixedCommand(ctx context.Context, dir string, extraEnv []string, name string, args ...string) error { startedAt := time.Now() - commandLine := redactedDistributionCommandLine(name, args) + commandLine := distributionCommandLine(name, args) log.Printf("RUN phase=distribution_build.command status=starting workdir=%s command=%s envKeys=%s", safeOptional(dir), commandLine, envKeysSummary(extraEnvMap(extraEnv), nil)) command := exec.CommandContext(ctx, name, args...) command.Dir = dir @@ -493,26 +413,16 @@ func fixedCommand(ctx context.Context, dir string, extraEnv []string, name strin command.Stdout = io.Discard command.Stderr = io.Discard if err := command.Run(); err != nil { - log.Printf("RUN phase=distribution_build.command status=failed command=%s durationMs=%d error=%s", commandLine, time.Since(startedAt).Milliseconds(), RedactText(err.Error())) + log.Printf("RUN phase=distribution_build.command status=failed command=%s durationMs=%d error=%s", commandLine, time.Since(startedAt).Milliseconds(), err.Error()) return err } log.Printf("RUN phase=distribution_build.command status=complete command=%s durationMs=%d", commandLine, time.Since(startedAt).Milliseconds()) return nil } -func redactedDistributionCommandLine(name string, args []string) string { +func distributionCommandLine(name string, args []string) string { parts := append([]string{name}, args...) - redacted := append([]string(nil), parts...) - for index, part := range redacted { - if part == "-ldflags" && index+1 < len(redacted) { - redacted[index+1] = "[redacted-ldflags]" - continue - } - if strings.Contains(part, "BuildRegistrationToken=") { - redacted[index] = "[redacted-ldflags]" - } - } - return redactedCommandLine(redacted) + return quotedCommandLine(parts) } func extraEnvMap(entries []string) map[string]string { @@ -531,104 +441,6 @@ func extraEnvMap(entries []string) map[string]string { return env } -func createDistributionArchive(path string, format string, binaryPath string, configPath string) error { - if format == "zip" { - file, err := os.Create(path) - if err != nil { - return err - } - writer := zip.NewWriter(file) - if err := addZipFile(writer, binaryPath); err != nil { - writer.Close() - file.Close() - return err - } - if err := addZipFile(writer, configPath); err != nil { - writer.Close() - file.Close() - return err - } - if err := writer.Close(); err != nil { - file.Close() - return err - } - return file.Close() - } - file, err := os.Create(path) - if err != nil { - return err - } - gzipWriter := gzip.NewWriter(file) - tarWriter := tar.NewWriter(gzipWriter) - if err := addTarFile(tarWriter, binaryPath); err != nil { - tarWriter.Close() - gzipWriter.Close() - file.Close() - return err - } - if err := addTarFile(tarWriter, configPath); err != nil { - tarWriter.Close() - gzipWriter.Close() - file.Close() - return err - } - if err := tarWriter.Close(); err != nil { - gzipWriter.Close() - file.Close() - return err - } - if err := gzipWriter.Close(); err != nil { - file.Close() - return err - } - return file.Close() -} - -func addZipFile(writer *zip.Writer, path string) error { - body, err := os.ReadFile(path) - if err != nil { - return err - } - entry, err := writer.Create(filepath.Base(path)) - if err != nil { - return err - } - _, err = entry.Write(body) - return err -} - -func addTarFile(writer *tar.Writer, path string) error { - info, err := os.Stat(path) - if err != nil { - return err - } - header := &tar.Header{Name: filepath.Base(path), Mode: 0o600, Size: info.Size()} - if strings.HasSuffix(filepath.Base(path), ".exe") || filepath.Base(path) == "run" { - header.Mode = 0o700 - } - if err := writer.WriteHeader(header); err != nil { - return err - } - file, err := os.Open(path) - if err != nil { - return err - } - defer file.Close() - _, err = io.Copy(writer, file) - return err -} - -func archiveFilename(input protocol.DistributionBuildInputResponse) string { - base := "run-" + input.ServerInstanceID - if input.ComponentKind == "client-manager" { - base = input.ProfileKey + "-" + input.ServerInstanceID - } - if input.PackageFormat == "zip" { - return base + ".zip" - } - return base + ".tar.gz" -} - func distributionBuildWorkspace(workspaceRoot string, pluginID string, jobID string) string { return filepath.Join(workspaceRoot, "distribution-builds", safeWorkspaceName(pluginID), safeWorkspaceName(jobID)) } diff --git a/runtime/distribution_build_test.go b/runtime/distribution_build_test.go index fe6bf87..2b506ee 100644 --- a/runtime/distribution_build_test.go +++ b/runtime/distribution_build_test.go @@ -1,14 +1,10 @@ package runtime import ( - "archive/tar" - "archive/zip" "bytes" - "compress/gzip" "context" "encoding/base64" "encoding/json" - "io" "os" "os/exec" "path/filepath" @@ -181,13 +177,22 @@ func TestDistributionBuildIsolationUsesPluginJobWorkspaceAndDistinctPackageState worker := &Worker{cfg: workerTestConfig(t), client: client} worker.state.RunEndpointID = "run-test" worker.state.SessionToken = "session-token" + artifactDir := t.TempDir() + firstArtifactPath := filepath.Join(artifactDir, "alpha.bin") + secondArtifactPath := filepath.Join(artifactDir, "beta.bin") + if err := os.WriteFile(firstArtifactPath, []byte("alpha archive"), 0o600); err != nil { + t.Fatalf("write first artifact: %v", err) + } + if err := os.WriteFile(secondArtifactPath, []byte("beta archive"), 0o600); err != nil { + t.Fatalf("write second artifact: %v", err) + } client.buildInput = firstInput - if err := worker.uploadDistributionArtifact(context.Background(), firstAssignment, firstInput.ArtifactID, []byte("alpha archive")); err != nil { + if err := worker.uploadDistributionArtifact(context.Background(), firstAssignment, firstInput.ArtifactID, firstArtifactPath); err != nil { t.Fatalf("upload first artifact: %v", err) } client.artifactPayload = nil client.buildInput = secondInput - if err := worker.uploadDistributionArtifact(context.Background(), secondAssignment, secondInput.ArtifactID, []byte("beta archive")); err != nil { + if err := worker.uploadDistributionArtifact(context.Background(), secondAssignment, secondInput.ArtifactID, secondArtifactPath); err != nil { t.Fatalf("upload second artifact: %v", err) } if len(client.artifactOpenRequests) != 2 { @@ -201,38 +206,6 @@ func TestDistributionBuildIsolationUsesPluginJobWorkspaceAndDistinctPackageState } } -func TestCreateDistributionArchiveIncludesExecutableAndConfigForSupportedFormats(t *testing.T) { - for _, format := range []string{"tar.gz", "zip"} { - t.Run(format, func(t *testing.T) { - root := t.TempDir() - executableName := "run" - if format == "zip" { - executableName = "run.exe" - } - binaryPath := filepath.Join(root, executableName) - configPath := filepath.Join(root, "config.json") - if err := os.WriteFile(binaryPath, []byte("binary"), 0o700); err != nil { - t.Fatalf("write binary: %v", err) - } - if err := os.WriteFile(configPath, []byte(`{"kind":"run"}`), 0o600); err != nil { - t.Fatalf("write config: %v", err) - } - archivePath := filepath.Join(root, "package."+strings.ReplaceAll(format, ".", "")) - if err := createDistributionArchive(archivePath, format, binaryPath, configPath); err != nil { - t.Fatalf("create archive: %v", err) - } - payload, err := os.ReadFile(archivePath) - if err != nil { - t.Fatalf("read archive: %v", err) - } - entries := archiveEntries(t, format, payload) - if !entries[executableName] || !entries["config.json"] { - t.Fatalf("expected executable and config in %s archive, got %+v", format, entries) - } - }) - } -} - func TestPrepareDistributionSourceCopiesTrustedRunSourceIntoWorkspace(t *testing.T) { sourceRoot := t.TempDir() if err := os.WriteFile(filepath.Join(sourceRoot, "go.mod"), []byte("module example.test/trusted\n\ngo 1.24\n"), 0o600); err != nil { @@ -269,23 +242,14 @@ func TestPrepareDistributionSourceCopiesTrustedRunSourceIntoWorkspace(t *testing } } -func TestValidateDistributionBuildInputRejectsUnapprovedClientSource(t *testing.T) { +func TestValidateDistributionBuildInputRejectsLegacyClientManagerBuilds(t *testing.T) { assignment := protocol.RunJobAssignment{JobID: "job-build", ServerInstanceID: "server-build", RunEndpointID: "run-build"} - base := protocol.DistributionBuildInputResponse{ + input := protocol.DistributionBuildInputResponse{ JobID: assignment.JobID, ComponentKind: "client-manager", ServerInstanceID: assignment.ServerInstanceID, RunEndpointID: assignment.RunEndpointID, PluginID: "game.scum", TargetOS: "linux", TargetArch: "amd64", PackageFormat: "tar.gz", ArtifactID: "artifact-build", OutputFilename: "manager", AuthKey: "key", SourceRevision: "main", } - for _, repository := range []string{"http://example.test/manager.git", "https://token@example.test/manager.git", "https://example.test/manager.git?ref=main"} { - input := base - input.RepositoryURL = repository - if err := validateDistributionBuildInput(assignment, input); err == nil { - t.Fatalf("expected repository %q to be rejected", repository) - } - } - base.RepositoryURL = "https://example.test/manager.git" - base.SourceRevision = "" - if err := validateDistributionBuildInput(assignment, base); err == nil { - t.Fatal("expected an unpinned client-manager source to be rejected") + if err := validateDistributionBuildInput(assignment, input); err == nil || !strings.Contains(err.Error(), "component kind") { + t.Fatalf("expected legacy component kind to be rejected, got %v", err) } } @@ -301,99 +265,7 @@ func TestValidateDistributionBuildInputAllowsDedicatedRunIdentity(t *testing.T) } input.WorkspaceSeed = "" input.ComponentKind = "client-manager" - input.PackageFormat = "zip" - input.RepositoryURL = "https://example.test/manager.git" - input.SourceRevision = "main" - if err := validateDistributionBuildInput(assignment, input); err == nil { - t.Fatal("client-manager build must remain bound to its assigned builder") + if err := validateDistributionBuildInput(assignment, input); err == nil || !strings.Contains(err.Error(), "component kind") { + t.Fatalf("expected client-manager build to be rejected, got %v", err) } } - -func archiveEntries(t *testing.T, format string, payload []byte) map[string]bool { - t.Helper() - entries := map[string]bool{} - if format == "zip" { - reader, err := zip.NewReader(bytes.NewReader(payload), int64(len(payload))) - if err != nil { - t.Fatalf("open zip: %v", err) - } - for _, file := range reader.File { - entries[file.Name] = true - } - return entries - } - gzipReader, err := gzip.NewReader(bytes.NewReader(payload)) - if err != nil { - t.Fatalf("open gzip: %v", err) - } - defer gzipReader.Close() - reader := tar.NewReader(gzipReader) - for { - header, err := reader.Next() - if err == io.EOF { - break - } - if err != nil { - t.Fatalf("read tar: %v", err) - } - entries[header.Name] = true - } - return entries -} - -func extractArchive(t *testing.T, format string, payload []byte, destination string) { - t.Helper() - if format == "zip" { - reader, err := zip.NewReader(bytes.NewReader(payload), int64(len(payload))) - if err != nil { - t.Fatalf("open zip: %v", err) - } - for _, file := range reader.File { - input, err := file.Open() - if err != nil { - t.Fatalf("open zip entry: %v", err) - } - body, err := io.ReadAll(input) - closeErr := input.Close() - if err != nil || closeErr != nil { - t.Fatalf("read zip entry: err=%v close=%v", err, closeErr) - } - mode := os.FileMode(0o600) - if file.Name == "run" || strings.HasSuffix(file.Name, ".exe") { - mode = 0o700 - } - if err := os.WriteFile(filepath.Join(destination, file.Name), body, mode); err != nil { - t.Fatalf("write zip entry: %v", err) - } - } - return - } - gzipReader, err := gzip.NewReader(bytes.NewReader(payload)) - if err != nil { - t.Fatalf("open gzip: %v", err) - } - defer gzipReader.Close() - reader := tar.NewReader(gzipReader) - for { - header, err := reader.Next() - if err == io.EOF { - break - } - if err != nil { - t.Fatalf("read tar: %v", err) - } - mode := os.FileMode(header.Mode) - if err := os.WriteFile(filepath.Join(destination, header.Name), mustReadAll(t, reader), mode); err != nil { - t.Fatalf("write tar entry: %v", err) - } - } -} - -func mustReadAll(t *testing.T, reader io.Reader) []byte { - t.Helper() - body, err := io.ReadAll(reader) - if err != nil { - t.Fatalf("read archive entry: %v", err) - } - return body -} diff --git a/runtime/file_artifact_transfer.go b/runtime/file_artifact_transfer.go index 2bc8fcf..90c5fc9 100644 --- a/runtime/file_artifact_transfer.go +++ b/runtime/file_artifact_transfer.go @@ -60,7 +60,7 @@ func (worker *Worker) uploadFileArtifact(ctx context.Context, assignment protoco if read != length { return fmt.Errorf("file artifact chunk is shorter than expected") } - chunk := append([]byte(nil), buffer[:length]...) + chunk := buffer[:length] state, err := worker.registeredState() if err != nil { return err diff --git a/runtime/lifecycle.go b/runtime/lifecycle.go index 9298216..f89569c 100644 --- a/runtime/lifecycle.go +++ b/runtime/lifecycle.go @@ -338,7 +338,7 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment if len(assignment.ExecutionInput.DLLExtensions) > 0 { log.Printf("RUN phase=lifecycle.dll status=validating job=%s extensions=%d", assignment.JobID, len(assignment.ExecutionInput.DLLExtensions)) if err := protocol.ValidateRunJobAssignment(assignment); err != nil { - log.Printf("RUN phase=lifecycle.dll status=invalid job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.dll status=invalid job=%s error=%s", assignment.JobID, err.Error()) return lifecycleFailure("unsafe_dll_extension_plan", "DLL extension plan is invalid") } } @@ -353,7 +353,7 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment log.Printf("RUN phase=lifecycle.template status=loading job=%s target=%s", assignment.JobID, safeOptional(assignment.TargetKey)) template, scope, err := executor.loadLifecycleTemplate(assignment) if err != nil { - log.Printf("RUN phase=lifecycle.template status=failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.template status=failed job=%s error=%s", assignment.JobID, err.Error()) return lifecycleFailure("unsafe_lifecycle_command", err.Error()) } log.Printf("RUN phase=lifecycle.template status=loaded job=%s action=%s mode=%s scope=%s commandArgs=%d envKeys=%s", assignment.JobID, safeOptional(template.Action), safeOptional(template.Mode), scope, len(template.Command)+len(template.Arguments), envKeysSummary(template.Env, template.Environment)) @@ -375,7 +375,7 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment if assignment.Capability == protocol.RunCapabilityProcessStart && len(assignment.ExecutionInput.DLLExtensions) > 0 { log.Printf("RUN phase=lifecycle.dll status=synchronizing job=%s extensions=%d", assignment.JobID, len(assignment.ExecutionInput.DLLExtensions)) if err := executor.synchronizeUE4SSDLLExtensions(ctx, assignment, template, scope); err != nil { - log.Printf("RUN phase=lifecycle.dll status=failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.dll status=failed job=%s error=%s", assignment.JobID, err.Error()) return dllExtensionLifecycleFailure(err) } log.Printf("RUN phase=lifecycle.dll status=complete job=%s", assignment.JobID) @@ -387,16 +387,16 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment log.Printf("RUN phase=lifecycle.command status=building job=%s", assignment.JobID) command, err := template.ToProcessCommand(scope, NewWorkspaceResolver(executor.workspaceRoot), assignment) if err != nil { - log.Printf("RUN phase=lifecycle.command status=build_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.command status=build_failed job=%s error=%s", assignment.JobID, err.Error()) return lifecycleFailure("unsafe_lifecycle_command", err.Error()) } - log.Printf("RUN phase=lifecycle.command status=starting job=%s workdir=%s command=%s timeoutMs=%d envKeys=%s", assignment.JobID, safeOptional(command.WorkDir), redactedCommandLine(command.Args), command.Timeout.Milliseconds(), envKeysSummary(command.Env, nil)) + log.Printf("RUN phase=lifecycle.command status=starting job=%s workdir=%s command=%s timeoutMs=%d envKeys=%s", assignment.JobID, safeOptional(command.WorkDir), quotedCommandLine(command.Args), command.Timeout.Milliseconds(), envKeysSummary(command.Env, nil)) command.OutputLine = func(stream string, line string) { _ = executor.logSink.Append(ctx, assignment, stream, line) } result, err := executor.supervisor.Run(ctx, command) if err != nil && ctx.Err() != nil { - log.Printf("RUN phase=lifecycle.command status=cancelled job=%s error=%s", assignment.JobID, RedactText(ctx.Err().Error())) + log.Printf("RUN phase=lifecycle.command status=cancelled job=%s error=%s", assignment.JobID, ctx.Err().Error()) return LifecycleExecutionResult{ State: lifecycleResultStateCancelled, Progress: protocol.RunJobProgressReport{Percent: 100, Message: "lifecycle action cancelled"}, @@ -408,8 +408,8 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment executor.writeProcessLogs(ctx, assignment, result) } if err != nil { - log.Printf("RUN phase=lifecycle.command status=failed job=%s exitCode=%d error=%s", assignment.JobID, result.ExitCode, RedactText(err.Error())) - return lifecycleFailure("lifecycle_process_failed", RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.command status=failed job=%s exitCode=%d error=%s", assignment.JobID, result.ExitCode, err.Error()) + return lifecycleFailure("lifecycle_process_failed", err.Error()) } if result.ExitCode != 0 { log.Printf("RUN phase=lifecycle.command status=failed job=%s exitCode=%d", assignment.JobID, result.ExitCode) @@ -418,7 +418,7 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment log.Printf("RUN phase=lifecycle.command status=exited job=%s exitCode=%d stdoutBytes=%d stderrBytes=%d", assignment.JobID, result.ExitCode, len(result.Stdout), len(result.Stderr)) artifactRef, err := executor.artifactHook.QueueLifecycleResult(ctx, assignment, result) if err != nil { - log.Printf("RUN phase=lifecycle.artifact status=failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.artifact status=failed job=%s error=%s", assignment.JobID, err.Error()) return lifecycleFailure("lifecycle_artifact_hook_failed", err.Error()) } log.Printf("RUN phase=lifecycle status=succeeded job=%s resultRef=%s", assignment.JobID, safeOptional(artifactRef)) @@ -468,11 +468,11 @@ func (executor LifecycleExecutor) executeDeployment(ctx context.Context, assignm return lifecycleFailure("deployment_shell_unsupported", "deployment shell is not supported by this Run") } command := ProcessCommand{Args: args, WorkDir: workdir, JobID: assignment.JobID, Capability: assignment.Capability, Action: action} - log.Printf("RUN phase=deployment.command status=starting job=%s action=%s revision=%d root=%s workdir=%s command=%s", assignment.JobID, action, definition.Revision, safeOptional(definition.ServerRoot), safeOptional(workdir), redactedCommandLine(command.Args)) + log.Printf("RUN phase=deployment.command status=starting job=%s action=%s revision=%d root=%s workdir=%s command=%s", assignment.JobID, action, definition.Revision, safeOptional(definition.ServerRoot), safeOptional(workdir), quotedCommandLine(command.Args)) result, err := executor.supervisor.Run(ctx, command) if err != nil { - log.Printf("RUN phase=deployment.command status=failed job=%s exitCode=%d error=%s", assignment.JobID, result.ExitCode, RedactText(err.Error())) - return lifecycleFailure("lifecycle_process_failed", RedactText(err.Error())) + log.Printf("RUN phase=deployment.command status=failed job=%s exitCode=%d error=%s", assignment.JobID, result.ExitCode, err.Error()) + return lifecycleFailure("lifecycle_process_failed", err.Error()) } executor.writeProcessLogs(ctx, assignment, result) if result.ExitCode != 0 { @@ -590,18 +590,18 @@ func (executor LifecycleExecutor) executeManaged(ctx context.Context, assignment log.Printf("RUN phase=lifecycle.managed status=building_start_command job=%s scope=%s", assignment.JobID, scope) command, err := template.ToManagedProcessCommand(resolver, scope, assignment) if err != nil { - log.Printf("RUN phase=lifecycle.managed status=build_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.managed status=build_failed job=%s error=%s", assignment.JobID, err.Error()) return lifecycleFailure("unsafe_lifecycle_command", err.Error()) } - log.Printf("RUN phase=lifecycle.managed status=starting job=%s workdir=%s command=%s timeoutMs=%d envKeys=%s", assignment.JobID, safeOptional(command.WorkDir), redactedCommandLine(command.Args), command.Timeout.Milliseconds(), envKeysSummary(command.Env, nil)) + log.Printf("RUN phase=lifecycle.managed status=starting job=%s workdir=%s command=%s timeoutMs=%d envKeys=%s", assignment.JobID, safeOptional(command.WorkDir), quotedCommandLine(command.Args), command.Timeout.Milliseconds(), envKeysSummary(command.Env, nil)) item, err := executor.managed.Start(ctx, command, identity, executor.managedProcessOutput(ctx, assignment)) if err != nil { if ctx.Err() != nil { - log.Printf("RUN phase=lifecycle.managed status=cancelled job=%s error=%s", assignment.JobID, RedactText(ctx.Err().Error())) + log.Printf("RUN phase=lifecycle.managed status=cancelled job=%s error=%s", assignment.JobID, ctx.Err().Error()) return lifecycleExecutionFailure("lifecycle_cancelled", "lifecycle action cancelled", false) } - log.Printf("RUN phase=lifecycle.managed status=failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) - return lifecycleFailure("lifecycle_process_failed", RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.managed status=failed job=%s error=%s", assignment.JobID, err.Error()) + return lifecycleFailure("lifecycle_process_failed", err.Error()) } log.Printf("RUN phase=lifecycle.managed status=started job=%s pid=%d state=%s stdoutRef=%s stderrRef=%s", assignment.JobID, item.PID, item.State, safeOptional(item.StdoutLogRef), safeOptional(item.StderrLogRef)) return processExecutionResult(item, "process started") @@ -610,8 +610,8 @@ func (executor LifecycleExecutor) executeManaged(ctx context.Context, assignment log.Printf("RUN phase=lifecycle.managed status=stopping job=%s scope=%s", assignment.JobID, scope) item, err := executor.managed.Stop(ctx, identity) if err != nil { - log.Printf("RUN phase=lifecycle.managed status=stop_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) - return lifecycleFailure("lifecycle_stop_failed", RedactText(err.Error())) + log.Printf("RUN phase=lifecycle.managed status=stop_failed job=%s error=%s", assignment.JobID, err.Error()) + return lifecycleFailure("lifecycle_stop_failed", err.Error()) } log.Printf("RUN phase=lifecycle.managed status=stopped job=%s pid=%d state=%s classification=%s", assignment.JobID, item.PID, item.State, safeOptional(item.ExitClassification)) return processExecutionResult(item, "process stopped") @@ -695,7 +695,7 @@ func processExecutionResult(item ProcessIdentity, message string) LifecycleExecu } func lifecycleExecutionFailure(code string, message string, retryable bool) LifecycleExecutionResult { - return LifecycleExecutionResult{State: lifecycleResultStateFailed, Progress: protocol.RunJobProgressReport{Percent: 100, Message: RedactText(message)}, Message: RedactText(message), ErrorCode: code, Retryable: retryable, ExecutionResult: protocol.RunJobExecutionResult{Kind: "file", Summary: code}} + return LifecycleExecutionResult{State: lifecycleResultStateFailed, Progress: protocol.RunJobProgressReport{Percent: 100, Message: message}, Message: message, ErrorCode: code, Retryable: retryable, ExecutionResult: protocol.RunJobExecutionResult{Kind: "file", Summary: code}} } func (template LifecycleActionTemplate) ToProcessCommand(workdir string, resolver WorkspaceResolver, assignment protocol.RunJobAssignment) (ProcessCommand, error) { @@ -948,7 +948,7 @@ func (supervisor OSProcessSupervisor) Run(ctx context.Context, command ProcessCo return ProcessResult{ExitCode: -1}, fmt.Errorf("command is required") } startedAt := time.Now() - log.Printf("RUN phase=process.command status=starting job=%s capability=%s action=%s workdir=%s command=%s timeoutMs=%d envKeys=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), safeOptional(command.WorkDir), redactedCommandLine(command.Args), command.Timeout.Milliseconds(), envKeysSummary(command.Env, nil)) + log.Printf("RUN phase=process.command status=starting job=%s capability=%s action=%s workdir=%s command=%s timeoutMs=%d envKeys=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), safeOptional(command.WorkDir), quotedCommandLine(command.Args), command.Timeout.Milliseconds(), envKeysSummary(command.Env, nil)) if command.Timeout > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, command.Timeout) @@ -967,14 +967,14 @@ func (supervisor OSProcessSupervisor) Run(ctx context.Context, command ProcessCo cmd.Stdout = stdoutWriter cmd.Stderr = stderrWriter if err := cmd.Start(); err != nil { - log.Printf("RUN phase=process.command status=start_failed job=%s capability=%s action=%s command=%s durationMs=%d error=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), redactedCommandLine(command.Args), time.Since(startedAt).Milliseconds(), RedactText(err.Error())) + log.Printf("RUN phase=process.command status=start_failed job=%s capability=%s action=%s command=%s durationMs=%d error=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), quotedCommandLine(command.Args), time.Since(startedAt).Milliseconds(), err.Error()) return ProcessResult{ExitCode: -1}, err } pid := 0 if cmd.Process != nil { pid = cmd.Process.Pid } - log.Printf("RUN phase=process.command status=started job=%s capability=%s action=%s pid=%d command=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), pid, redactedCommandLine(command.Args)) + log.Printf("RUN phase=process.command status=started job=%s capability=%s action=%s pid=%d command=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), pid, quotedCommandLine(command.Args)) err := cmd.Wait() stdoutWriter.Flush() stderrWriter.Flush() @@ -983,10 +983,10 @@ func (supervisor OSProcessSupervisor) Run(ctx context.Context, command ProcessCo result.ExitCode = cmd.ProcessState.ExitCode() } if err != nil { - log.Printf("RUN phase=process.command status=failed job=%s capability=%s action=%s pid=%d command=%s exitCode=%d durationMs=%d stdoutBytes=%d stderrBytes=%d error=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), pid, redactedCommandLine(command.Args), result.ExitCode, time.Since(startedAt).Milliseconds(), len(result.Stdout), len(result.Stderr), RedactText(err.Error())) + log.Printf("RUN phase=process.command status=failed job=%s capability=%s action=%s pid=%d command=%s exitCode=%d durationMs=%d stdoutBytes=%d stderrBytes=%d error=%s", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), pid, quotedCommandLine(command.Args), result.ExitCode, time.Since(startedAt).Milliseconds(), len(result.Stdout), len(result.Stderr), err.Error()) return result, err } - log.Printf("RUN phase=process.command status=exited job=%s capability=%s action=%s pid=%d command=%s exitCode=%d durationMs=%d stdoutBytes=%d stderrBytes=%d", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), pid, redactedCommandLine(command.Args), result.ExitCode, time.Since(startedAt).Milliseconds(), len(result.Stdout), len(result.Stderr)) + log.Printf("RUN phase=process.command status=exited job=%s capability=%s action=%s pid=%d command=%s exitCode=%d durationMs=%d stdoutBytes=%d stderrBytes=%d", safeOptional(command.JobID), safeOptional(command.Capability), safeOptional(command.Action), pid, quotedCommandLine(command.Args), result.ExitCode, time.Since(startedAt).Milliseconds(), len(result.Stdout), len(result.Stderr)) return result, nil } @@ -1127,8 +1127,8 @@ func isSupportedRemoteCapability(capability string) bool { func lifecycleFailure(code string, message string) LifecycleExecutionResult { return LifecycleExecutionResult{ State: lifecycleResultStateFailed, - Progress: protocol.RunJobProgressReport{Percent: 100, Message: RedactText(message)}, - Message: RedactText(message), + Progress: protocol.RunJobProgressReport{Percent: 100, Message: message}, + Message: message, ErrorCode: code, } } @@ -1183,23 +1183,23 @@ func safeOptional(value string) string { if value == "" { return "-" } - return RedactText(value) + return value } func errorSummary(err error) string { if err == nil { return "-" } - return RedactText(err.Error()) + return err.Error() } -func redactedCommandLine(args []string) string { +func quotedCommandLine(args []string) string { if len(args) == 0 { return "-" } parts := make([]string, len(args)) for i, arg := range args { - parts[i] = strconv.Quote(RedactText(arg)) + parts[i] = strconv.Quote(arg) } return strings.Join(parts, " ") } @@ -1223,18 +1223,6 @@ func envKeysSummary(first map[string]string, second map[string]string) string { return strings.Join(keys, ",") } -func RedactText(value string) string { - redacted := value - replacements := []string{"/Users/", "[host]/", "Bearer ", "Bearer [redacted] ", "sk-", "sk-[redacted]", "password=", "password=[redacted]", "api_key=", "api_key=[redacted]", "secret=", "secret=[redacted]", "unix://", "socket://"} - for i := 0; i+1 < len(replacements); i += 2 { - redacted = strings.ReplaceAll(redacted, replacements[i], replacements[i+1]) - } - if len(redacted) > maxLifecycleOutputBytes { - return redacted[:maxLifecycleOutputBytes] - } - return redacted -} - // splitRawLogLines only removes the newline framing used by LogEntry. It // deliberately preserves every other byte, including blank lines and spaces. func splitRawLogLines(value string) []string { diff --git a/runtime/lifecycle_test.go b/runtime/lifecycle_test.go index e766ec1..83030db 100644 --- a/runtime/lifecycle_test.go +++ b/runtime/lifecycle_test.go @@ -417,7 +417,6 @@ func TestResolveRuntimeProfilesSupportsDeclaredModesAndSafeMissingKeys(t *testin {Key: "run-local", Mode: RuntimeModeLocalProcess, Capabilities: []string{protocol.RunCapabilityProcessStart}, ActionRefs: map[string]string{"start": "actions/start.json"}, TransportKeys: []string{"server-files"}, Platforms: []string{"linux"}}, {Key: "hosted-ftp", Mode: RuntimeModeHostedFTPRCON, Capabilities: []string{protocol.RunCapabilityRemoteFTPRead, protocol.RunCapabilityRemoteRunRCONCommand}, TransportKeys: []string{"ftp", "rcon"}}, {Key: "ftp-only", Mode: RuntimeModeFTPOnly, Capabilities: []string{protocol.RunCapabilityRemoteFTPRead}, TransportKeys: []string{"ftp"}}, - {Key: "custom-client", Mode: RuntimeModeCustomClient, Capabilities: []string{protocol.RunCapabilityRemoteRunRCONCommand}, TransportKeys: []string{"rcon"}, ClientManagerRef: "scum-client-manager"}, }, LogSources: []RuntimeLogSource{{Key: "latest-log", Kind: "file.tail", TargetKey: "logs/latest", StreamKey: "latest-log"}}, TransportProfiles: []RuntimeTransportProfile{ @@ -426,23 +425,6 @@ func TestResolveRuntimeProfilesSupportsDeclaredModesAndSafeMissingKeys(t *testin {Key: "rcon", Kind: "rcon", TargetKey: "rcon", Capabilities: []string{protocol.RunCapabilityRemoteRunRCONCommand}}, }, } - resolution, err := ResolveRuntimeProfile(profiles, "custom-client", "windows", RuntimeBindingSet{ - ProfileKey: "custom-client", - Mode: RuntimeModeCustomClient, - Bindings: map[string]string{ - "rcon": "binding://rcon/current", - "logs/latest": "binding://logs/latest", - "steamcmd": "binding://probe/steamcmd", - "scum-client-manager": "binding://client/current", - }, - }) - if err != nil { - t.Fatalf("resolve custom client profile: %v", err) - } - if !resolution.Available || resolution.Mode != RuntimeModeCustomClient || resolution.ClientManagerRef != "scum-client-manager" { - t.Fatalf("unexpected custom client resolution: %+v", resolution) - } - missing, err := ResolveRuntimeProfile(profiles, "hosted-ftp", "linux", RuntimeBindingSet{ProfileKey: "hosted-ftp", Mode: RuntimeModeHostedFTPRCON, Bindings: map[string]string{"ftp-root": "binding://ftp/current"}}) if err != nil { t.Fatalf("resolve hosted profile: %v", err) @@ -476,7 +458,7 @@ func TestTailDeclaredFileLogSourceUsesCheckpointAndVerbatimOutput(t *testing.T) t.Fatalf("expected verbatim tailed lines, got %+v", sink.lines) } checkpoint := store.GetLogCheckpoint("latest-log") - if checkpoint.Offset == 0 || checkpoint.Sequence != 2 || strings.Contains(RedactedLogCheckpointSummary(checkpoint), "/Users/") { + if checkpoint.Offset == 0 || checkpoint.Sequence != 2 || strings.Contains(LogCheckpointSummary(checkpoint), "/Users/") { t.Fatalf("expected durable safe checkpoint, got %+v", checkpoint) } @@ -508,14 +490,18 @@ func TestTailDeclaredFileLogSourceDoesNotLimitOrRewriteOutput(t *testing.T) { result := TailDeclaredFileLogSource(context.Background(), root, assignment, source, sink, store) - if result.State != lifecycleResultStateSucceeded || len(sink.lines) != 3 { + if result.State != lifecycleResultStateSucceeded || len(sink.lines) < 3 { t.Fatalf("expected all unbounded log entries, result=%+v lines=%d", result, len(sink.lines)) } - if sink.lines[0] != "current-log:"+longLine || sink.lines[1] != "current-log:" || sink.lines[2] != "current-log:final" { - t.Fatalf("expected byte-for-byte log payloads, got lengths=%d,%d,%d", len(sink.lines[0]), len(sink.lines[1]), len(sink.lines[2])) + payloads := make([]string, len(sink.lines)) + for index, line := range sink.lines { + payloads[index] = strings.TrimPrefix(line, "current-log:") + } + if strings.Join(payloads[:len(payloads)-2], "") != longLine || payloads[len(payloads)-2] != "" || payloads[len(payloads)-1] != "final" { + t.Fatalf("expected byte-for-byte log payloads after chunk reassembly, got %d entries", len(payloads)) } checkpoint := store.GetLogCheckpoint(source.Key) - if checkpoint.Offset != int64(len(longLine+"\n\nfinal")) || checkpoint.Sequence != 3 { + if checkpoint.Offset != int64(len(longLine+"\n\nfinal")) || checkpoint.Sequence != uint64(len(payloads)) { t.Fatalf("expected complete checkpoint after unbounded tail, got %+v", checkpoint) } } diff --git a/runtime/log_sources.go b/runtime/log_sources.go index 0c114d8..af28fa0 100644 --- a/runtime/log_sources.go +++ b/runtime/log_sources.go @@ -82,14 +82,14 @@ func TailDeclaredFileLogSource(ctx context.Context, workspaceRoot string, assign return lifecycleFailure("log_source_seek_failed", err.Error()) } } - reader := bufio.NewReader(file) + reader := bufio.NewReaderSize(file, 64*1024) for { // A newline only frames an entry. Every other byte, including CR, blank // lines, and arbitrarily long output, remains untouched. - line, readErr := reader.ReadString('\n') + line, readErr := reader.ReadSlice('\n') if len(line) > 0 { checkpoint.Sequence++ - if err := sink.Append(ctx, assignment, source.StreamKey, strings.TrimSuffix(line, "\n")); err != nil { + if err := sink.Append(ctx, assignment, source.StreamKey, strings.TrimSuffix(string(line), "\n")); err != nil { return lifecycleFailure("log_source_sink_failed", err.Error()) } checkpoint.SourceKey = source.Key @@ -97,7 +97,7 @@ func TailDeclaredFileLogSource(ctx context.Context, workspaceRoot string, assign checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID)) store.PutLogCheckpoint(checkpoint) } - if readErr == nil { + if readErr == nil || readErr == bufio.ErrBufferFull { continue } if readErr == io.EOF { @@ -116,7 +116,7 @@ func TailDeclaredFileLogSource(ctx context.Context, workspaceRoot string, assign } } -func RedactedLogCheckpointSummary(checkpoint LogSourceCheckpoint) string { +func LogCheckpointSummary(checkpoint LogSourceCheckpoint) string { return strings.Join([]string{ "source=" + checkpoint.SourceKey, fmt.Sprintf("offset=%d", checkpoint.Offset), diff --git a/runtime/metrics.go b/runtime/metrics.go index 073ab36..08fb16b 100644 --- a/runtime/metrics.go +++ b/runtime/metrics.go @@ -60,7 +60,7 @@ func WithMetricCollector(collector MetricCollector) LifecycleExecutorOption { func (worker *Worker) reportMetricsDegraded(ctx context.Context, trigger string) { if err := worker.ReportMetricsOnce(ctx); err != nil { - log.Printf("RUN phase=metrics status=degraded trigger=%s error=%s", safeOptional(trigger), RedactText(err.Error())) + log.Printf("RUN phase=metrics status=degraded trigger=%s error=%s", safeOptional(trigger), err.Error()) } } @@ -79,7 +79,7 @@ func (worker *Worker) ReportMetricsOnce(ctx context.Context) error { sample.MemoryPercent = utilization.MemoryPercent sample.DiskPercent = utilization.DiskPercent if collectErr != nil { - log.Printf("RUN phase=metrics status=utilization_unavailable error=%s", RedactText(collectErr.Error())) + log.Printf("RUN phase=metrics status=utilization_unavailable error=%s", collectErr.Error()) } } reportCtx, cancel := context.WithTimeout(ctx, metricReportTimeout) diff --git a/runtime/process_supervisor.go b/runtime/process_supervisor.go index 409c9a6..6b7877f 100644 --- a/runtime/process_supervisor.go +++ b/runtime/process_supervisor.go @@ -22,6 +22,7 @@ const ( managedProcessOutputPollInterval = 50 * time.Millisecond managedProcessOutputDrainDelay = 750 * time.Millisecond managedProcessOutputRetryDelay = 500 * time.Millisecond + managedProcessOutputReaderSize = 64 * 1024 ) type ProcessIdentity struct { @@ -143,7 +144,7 @@ func (supervisor *OSManagedProcessSupervisor) Start(ctx context.Context, command supervisor.mu.Lock() defer supervisor.mu.Unlock() key := identity.Scope - log.Printf("RUN phase=process.managed status=start_requested job=%s server=%s scope=%s command=%s workdir=%s", safeOptional(identity.JobID), identity.ServerInstanceID, safeOptional(identity.Scope), redactedCommandLine(command.Args), safeOptional(command.WorkDir)) + log.Printf("RUN phase=process.managed status=start_requested job=%s server=%s scope=%s command=%s workdir=%s", safeOptional(identity.JobID), identity.ServerInstanceID, safeOptional(identity.Scope), quotedCommandLine(command.Args), safeOptional(command.WorkDir)) if existing, ok := supervisor.items[key]; ok && existing.State == "running" && supervisor.isAlive(existing) { if existing.LogSessionID == "" { logSessionID, err := newManagedProcessLogSessionID() @@ -167,7 +168,7 @@ func (supervisor *OSManagedProcessSupervisor) Start(ctx context.Context, command return existing, nil } if err := ctx.Err(); err != nil { - log.Printf("RUN phase=process.managed status=context_done job=%s error=%s", safeOptional(identity.JobID), RedactText(err.Error())) + log.Printf("RUN phase=process.managed status=context_done job=%s error=%s", safeOptional(identity.JobID), err.Error()) return ProcessIdentity{}, err } if len(command.Args) == 0 { @@ -187,14 +188,14 @@ func (supervisor *OSManagedProcessSupervisor) Start(ctx context.Context, command } files, identity, err := supervisor.prepareOutputFilesLocked(identity, startedAt) if err != nil { - log.Printf("RUN phase=process.managed status=prepare_output_failed job=%s error=%s", safeOptional(identity.JobID), RedactText(err.Error())) + log.Printf("RUN phase=process.managed status=prepare_output_failed job=%s error=%s", safeOptional(identity.JobID), err.Error()) return ProcessIdentity{}, err } log.Printf("RUN phase=process.managed status=output_ready job=%s stdoutRef=%s stderrRef=%s", safeOptional(identity.JobID), safeOptional(identity.StdoutLogRef), safeOptional(identity.StderrLogRef)) process, err := startManagedProcess(command, files, identity.StopEventName) if err != nil { files.close() - log.Printf("RUN phase=process.managed status=start_failed job=%s error=%s", safeOptional(identity.JobID), RedactText(err.Error())) + log.Printf("RUN phase=process.managed status=start_failed job=%s error=%s", safeOptional(identity.JobID), err.Error()) return ProcessIdentity{}, err } identity.SupervisorPID = process.PID() @@ -214,7 +215,7 @@ func (supervisor *OSManagedProcessSupervisor) Start(ctx context.Context, command } else { delete(supervisor.items, key) } - log.Printf("RUN phase=process.managed status=persist_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), identity.PID, RedactText(err.Error())) + log.Printf("RUN phase=process.managed status=persist_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), identity.PID, err.Error()) return ProcessIdentity{}, err } supervisor.startTailersLocked(identity, output) @@ -241,7 +242,7 @@ func (supervisor *OSManagedProcessSupervisor) Stop(ctx context.Context, identity } log.Printf("RUN phase=process.managed status=stop_requested job=%s pid=%d scope=%s", safeOptional(identity.JobID), current.PID, safeOptional(identity.Scope)) if err := requestManagedProcessStop(current); err != nil { - log.Printf("RUN phase=process.managed status=stop_signal_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, RedactText(err.Error())) + log.Printf("RUN phase=process.managed status=stop_signal_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, err.Error()) } supervisor.mu.Unlock() deadline := time.NewTimer(2 * time.Second) @@ -264,11 +265,11 @@ func (supervisor *OSManagedProcessSupervisor) Stop(ctx context.Context, identity } select { case <-ctx.Done(): - log.Printf("RUN phase=process.managed status=stop_context_done job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, RedactText(ctx.Err().Error())) + log.Printf("RUN phase=process.managed status=stop_context_done job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, ctx.Err().Error()) return ProcessIdentity{}, ctx.Err() case <-deadline.C: if err := forceManagedProcessStop(current); err != nil { - log.Printf("RUN phase=process.managed status=forced_stop_signal_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, RedactText(err.Error())) + log.Printf("RUN phase=process.managed status=forced_stop_signal_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), current.PID, err.Error()) } current.State = "stopped" current.ExitClassification = "forced-stop" @@ -404,7 +405,7 @@ func (supervisor *OSManagedProcessSupervisor) wait(key string, process managedPr supervisor.drainTailersAfter(item, managedProcessOutputDrainDelay) } if err != nil { - log.Printf("RUN phase=process.managed status=%s job=%s pid=%d state=%s exitCode=%d classification=%s error=%s", processTerminalStatus(item.State), safeOptional(item.JobID), pid, item.State, item.ExitCode, item.ExitClassification, RedactText(err.Error())) + log.Printf("RUN phase=process.managed status=%s job=%s pid=%d state=%s exitCode=%d classification=%s error=%s", processTerminalStatus(item.State), safeOptional(item.JobID), pid, item.State, item.ExitCode, item.ExitClassification, err.Error()) return } log.Printf("RUN phase=process.managed status=%s job=%s pid=%d state=%s exitCode=%d classification=%s", processTerminalStatus(item.State), safeOptional(item.JobID), pid, item.State, item.ExitCode, item.ExitClassification) @@ -516,31 +517,31 @@ func (supervisor *OSManagedProcessSupervisor) tailOutput(ctx context.Context, ta defer supervisor.removeTailer(tailerID, tailer, identity) file, err := os.Open(path) if err != nil { - log.Printf("RUN phase=process.managed.output status=tail_open_failed job=%s pid=%d stream=%s path=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, safeOptional(path), RedactText(err.Error())) + log.Printf("RUN phase=process.managed.output status=tail_open_failed job=%s pid=%d stream=%s path=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, safeOptional(path), err.Error()) return } defer file.Close() if offset > 0 { if _, err := file.Seek(offset, io.SeekStart); err != nil { - log.Printf("RUN phase=process.managed.output status=tail_seek_failed job=%s pid=%d stream=%s path=%s offset=%d error=%s", safeOptional(identity.JobID), identity.PID, stream, safeOptional(path), offset, RedactText(err.Error())) + log.Printf("RUN phase=process.managed.output status=tail_seek_failed job=%s pid=%d stream=%s path=%s offset=%d error=%s", safeOptional(identity.JobID), identity.PID, stream, safeOptional(path), offset, err.Error()) return } } - reader := bufio.NewReader(file) + reader := bufio.NewReaderSize(file, managedProcessOutputReaderSize) defer func() { log.Printf("RUN phase=process.managed.output status=tail_stop job=%s pid=%d stream=%s offset=%d", safeOptional(identity.JobID), identity.PID, stream, offset) }() for { - line, err := reader.ReadString('\n') + line, err := reader.ReadSlice('\n') if len(line) > 0 { startOffset := offset endOffset := offset + int64(len(line)) - text := strings.TrimSuffix(line, "\n") + text := strings.TrimSuffix(string(line), "\n") for { if sinkErr := sink(identity, ManagedProcessLine{Text: text, StartOffset: startOffset, EndOffset: endOffset}); sinkErr == nil { break } else { - log.Printf("RUN phase=process.managed.output status=sink_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, RedactText(sinkErr.Error())) + log.Printf("RUN phase=process.managed.output status=sink_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, sinkErr.Error()) } select { case <-ctx.Done(): @@ -552,7 +553,7 @@ func (supervisor *OSManagedProcessSupervisor) tailOutput(ctx context.Context, ta if offsetErr := supervisor.updateOutputOffset(identity, stream, endOffset); offsetErr == nil { break } else { - log.Printf("RUN phase=process.managed.output status=offset_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, RedactText(offsetErr.Error())) + log.Printf("RUN phase=process.managed.output status=offset_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, offsetErr.Error()) } select { case <-ctx.Done(): @@ -565,6 +566,9 @@ func (supervisor *OSManagedProcessSupervisor) tailOutput(ctx context.Context, ta if err == nil { continue } + if err == bufio.ErrBufferFull { + continue + } if err != io.EOF { return } @@ -589,7 +593,7 @@ func (supervisor *OSManagedProcessSupervisor) removeTailer(tailerID string, tail delete(supervisor.retired, key) if err := supervisor.persistLocked(); err != nil { supervisor.retired[key] = retired - log.Printf("RUN phase=process.managed.output status=retired_prune_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), identity.PID, RedactText(err.Error())) + log.Printf("RUN phase=process.managed.output status=retired_prune_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), identity.PID, err.Error()) } } } diff --git a/runtime/process_window_windows.go b/runtime/process_window_windows.go index 2824bc1..ece096c 100644 --- a/runtime/process_window_windows.go +++ b/runtime/process_window_windows.go @@ -52,17 +52,36 @@ type windowsFileManagedProcess struct { } type managedProcessHelperSpec struct { - Command ProcessCommand `json:"command"` - StopEventName string `json:"stopEventName,omitempty"` - StdoutPath string `json:"stdoutPath,omitempty"` - StderrPath string `json:"stderrPath,omitempty"` - PIDPath string `json:"pidPath,omitempty"` + Command managedProcessHelperCommand `json:"command"` + StopEventName string `json:"stopEventName,omitempty"` + StdoutPath string `json:"stdoutPath,omitempty"` + StderrPath string `json:"stderrPath,omitempty"` + PIDPath string `json:"pidPath,omitempty"` +} + +type managedProcessHelperCommand struct { + WorkDir string `json:"workDir,omitempty"` + Args []string `json:"args,omitempty"` + Env map[string]string `json:"env,omitempty"` + OutputMode string `json:"outputMode,omitempty"` + Timeout time.Duration `json:"timeout,omitempty"` + JobID string `json:"jobId,omitempty"` + Capability string `json:"capability,omitempty"` + Action string `json:"action,omitempty"` +} + +func managedProcessHelperCommandFromProcess(command ProcessCommand) managedProcessHelperCommand { + return managedProcessHelperCommand{WorkDir: command.WorkDir, Args: append([]string(nil), command.Args...), Env: copyStringMap(command.Env), OutputMode: command.OutputMode, Timeout: command.Timeout, JobID: command.JobID, Capability: command.Capability, Action: command.Action} +} + +func (command managedProcessHelperCommand) processCommand() ProcessCommand { + return ProcessCommand{WorkDir: command.WorkDir, Args: append([]string(nil), command.Args...), Env: copyStringMap(command.Env), OutputMode: command.OutputMode, Timeout: command.Timeout, JobID: command.JobID, Capability: command.Capability, Action: command.Action} } func startManagedProcess(command ProcessCommand, files managedProcessFiles, stopEventName string) (managedProcess, error) { pidPath := files.stdout.Name() + ".pid" _ = os.Remove(pidPath) - body, err := json.Marshal(managedProcessHelperSpec{Command: command, StopEventName: stopEventName, StdoutPath: files.stdout.Name(), StderrPath: files.stderr.Name(), PIDPath: pidPath}) + body, err := json.Marshal(managedProcessHelperSpec{Command: managedProcessHelperCommandFromProcess(command), StopEventName: stopEventName, StdoutPath: files.stdout.Name(), StderrPath: files.stderr.Name(), PIDPath: pidPath}) if err != nil { return nil, fmt.Errorf("encode managed process helper spec: %w", err) } @@ -189,7 +208,7 @@ func runManagedProcessHelper(spec managedProcessHelperSpec) (int, error) { return 1, fmt.Errorf("open managed process stderr: %w", err) } defer closeStderr() - command := spec.Command + command := spec.Command.processCommand() stopEventName := spec.StopEventName // The plugin-declared pipes mode uses ordinary inherited handles. The // durable output files are attached directly to the child process, so this diff --git a/runtime/runtime_profiles.go b/runtime/runtime_profiles.go index 4b8f876..7c06753 100644 --- a/runtime/runtime_profiles.go +++ b/runtime/runtime_profiles.go @@ -12,17 +12,15 @@ const ( RuntimeModeLocalProcess = "local-process" RuntimeModeHostedFTPRCON = "hosted-ftp-rcon" RuntimeModeFTPOnly = "ftp-only" - RuntimeModeCustomClient = "custom-client" ) type RuntimeProfiles struct { - Discovery []RuntimeDiscoveryProbe `json:"discovery,omitempty"` - LifecycleProfiles []RuntimeLifecycleProfile `json:"lifecycleProfiles,omitempty"` - DependencyProbes []RuntimeDependencyProbe `json:"dependencyProbes,omitempty"` - InstallPlans []RuntimeInstallPlan `json:"installPlans,omitempty"` - LogSources []RuntimeLogSource `json:"logSources,omitempty"` - TransportProfiles []RuntimeTransportProfile `json:"transportProfiles,omitempty"` - ClientManagers []RuntimeClientManagerSpec `json:"clientManagers,omitempty"` + Discovery []RuntimeDiscoveryProbe `json:"discovery,omitempty"` + LifecycleProfiles []RuntimeLifecycleProfile `json:"lifecycleProfiles,omitempty"` + DependencyProbes []RuntimeDependencyProbe `json:"dependencyProbes,omitempty"` + InstallPlans []RuntimeInstallPlan `json:"installPlans,omitempty"` + LogSources []RuntimeLogSource `json:"logSources,omitempty"` + TransportProfiles []RuntimeTransportProfile `json:"transportProfiles,omitempty"` } type RuntimeDiscoveryProbe struct { @@ -39,7 +37,6 @@ type RuntimeLifecycleProfile struct { Capabilities []string `json:"capabilities"` ActionRefs map[string]string `json:"actionRefs,omitempty"` TransportKeys []string `json:"transportKeys,omitempty"` - ClientManagerRef string `json:"clientManagerRef,omitempty"` Platforms []string `json:"platforms,omitempty"` } @@ -84,10 +81,6 @@ type RuntimeTransportProfile struct { Capabilities []string `json:"capabilities"` } -type RuntimeClientManagerSpec struct { - Key string `json:"key"` -} - type RuntimeBindingSet struct { ProfileKey string `json:"profileKey"` Mode string `json:"mode"` @@ -104,7 +97,6 @@ type RuntimeResolution struct { Transports []RuntimeTransportProfile `json:"transports,omitempty"` LogSources []RuntimeLogSource `json:"logSources,omitempty"` Discovery []RuntimeDiscoveryProbe `json:"discovery,omitempty"` - ClientManagerRef string `json:"clientManagerRef,omitempty"` MissingKeys []string `json:"missingKeys,omitempty"` Available bool `json:"available"` } @@ -144,7 +136,6 @@ func ResolveRuntimeProfile(profiles RuntimeProfiles, profileKey string, targetOS Transports: transports, LogSources: safeLogSources(profiles.LogSources, targetOS), Discovery: safeDiscovery(profiles.Discovery, targetOS), - ClientManagerRef: profile.ClientManagerRef, MissingKeys: missing, Available: len(missing) == 0, }, nil @@ -173,9 +164,6 @@ func validateRuntimeProfile(profile RuntimeLifecycleProfile) error { return fmt.Errorf("runtime action ref is unsafe") } } - if profile.ClientManagerRef != "" && !protocol.ValidLogicalFileKey(profile.ClientManagerRef) { - return fmt.Errorf("client manager ref is unsafe") - } return nil } @@ -222,9 +210,6 @@ func missingRuntimeBindingKeys(profile RuntimeLifecycleProfile, transports []Run logTargets[source.TargetKey] = struct{}{} } } - if profile.ClientManagerRef != "" { - required[profile.ClientManagerRef] = struct{}{} - } for _, key := range binding.MissingKeys { if _, logTarget := logTargets[key]; logTarget { continue @@ -266,7 +251,7 @@ func safeLogSources(sources []RuntimeLogSource, targetOS string) []RuntimeLogSou func supportedRuntimeMode(mode string) bool { switch mode { - case RuntimeModeLocalProcess, RuntimeModeHostedFTPRCON, RuntimeModeFTPOnly, RuntimeModeCustomClient: + case RuntimeModeLocalProcess, RuntimeModeHostedFTPRCON, RuntimeModeFTPOnly: return true default: return false diff --git a/runtime/sqlite_schema_probe_test.go b/runtime/sqlite_schema_probe_test.go index 1175216..21ce747 100644 --- a/runtime/sqlite_schema_probe_test.go +++ b/runtime/sqlite_schema_probe_test.go @@ -63,7 +63,7 @@ func TestSQLiteSchemaProbeRejectsUnscopedOrUnsafeRequests(t *testing.T) { assignment.ExecutionInput.SQLiteSchemaProbe.Binding.DatabaseIdentity = "C:/host/path" result = NewSQLiteSchemaProbeExecutor(t.TempDir()).Execute(context.Background(), assignment) if result.ErrorCode != "invalid_request" || strings.Contains(result.Message, "C:/") { - t.Fatalf("expected redacted binding rejection, got %+v", result) + t.Fatalf("expected bounded binding rejection, got %+v", result) } } diff --git a/runtime/worker.go b/runtime/worker.go index 0a481f8..ab2e0a0 100644 --- a/runtime/worker.go +++ b/runtime/worker.go @@ -326,7 +326,7 @@ func (worker *Worker) registerUnlocked(ctx context.Context) error { Capacity: worker.capacityReportFor(state), }) if err != nil { - log.Printf("RUN phase=register status=failed endpoint=%s error=%s", worker.cfg.RunEndpointID, RedactText(err.Error())) + log.Printf("RUN phase=register status=failed endpoint=%s error=%s", worker.cfg.RunEndpointID, err.Error()) return err } if !response.Accepted || response.SessionToken == "" { @@ -399,10 +399,10 @@ func (worker *Worker) HeartbeatOnce(ctx context.Context) error { }) if err != nil { if sessionInvalidError(err) { - log.Printf("RUN phase=heartbeat status=session_invalid endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + log.Printf("RUN phase=heartbeat status=session_invalid endpoint=%s error=%s", state.RunEndpointID, err.Error()) return worker.reregisterAndReconcile(ctx, "heartbeat_session_invalid", state.SessionToken) } - log.Printf("RUN phase=heartbeat status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + log.Printf("RUN phase=heartbeat status=failed endpoint=%s error=%s", state.RunEndpointID, err.Error()) return err } if !response.Accepted { @@ -452,10 +452,10 @@ func (worker *Worker) claimAndRunOnce(ctx context.Context, waitSeconds int) (boo }) if err != nil { if sessionInvalidError(err) { - log.Printf("RUN phase=claim status=session_invalid endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + log.Printf("RUN phase=claim status=session_invalid endpoint=%s error=%s", state.RunEndpointID, err.Error()) return false, worker.reregisterAndReconcile(ctx, "claim_session_invalid", state.SessionToken) } - log.Printf("RUN phase=claim status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + log.Printf("RUN phase=claim status=failed endpoint=%s error=%s", state.RunEndpointID, err.Error()) return false, err } if !claim.Accepted || !claim.HasJob || claim.Job == nil { @@ -482,7 +482,7 @@ func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.Run } log.Printf("RUN phase=job status=journal_store job=%s capability=%s attempt=%d", assignment.JobID, assignment.Capability, assignment.Attempt) if err := worker.journal.Store(assignment); err != nil { - log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, err.Error()) return err } log.Printf("RUN phase=job status=ack_start job=%s capability=%s", assignment.JobID, assignment.Capability) @@ -495,7 +495,7 @@ func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.Run Message: "job accepted by run worker", }) if err != nil { - log.Printf("RUN phase=job status=ack_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=ack_failed job=%s error=%s", assignment.JobID, err.Error()) return err } assignment = ack.Job @@ -505,7 +505,7 @@ func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.Run } log.Printf("RUN phase=job status=ack_accepted job=%s attempt=%d", assignment.JobID, assignment.Attempt) if err := worker.journal.Store(assignment); err != nil { - log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, err.Error()) return err } progressSequence := worker.nextProgressSequence(assignment.ProgressSequence) @@ -524,7 +524,7 @@ func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.Run Sequence: progressSequence, }) if err != nil { - log.Printf("RUN phase=job status=progress_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=progress_failed job=%s error=%s", assignment.JobID, err.Error()) return err } if !progress.Accepted { @@ -534,13 +534,13 @@ func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.Run assignment = progress.Job log.Printf("RUN phase=job status=progress_accepted job=%s percent=%d", assignment.JobID, assignment.Progress.Percent) if err := worker.journal.Store(assignment); err != nil { - log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=journal_store_failed job=%s error=%s", assignment.JobID, err.Error()) return err } log.Printf("RUN phase=job status=execute_start job=%s capability=%s", assignment.JobID, assignment.Capability) execution, assignment, cancelledByPlatform, err := worker.executeWithJobPolling(ctx, assignment) if err != nil { - log.Printf("RUN phase=job status=execute_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=execute_failed job=%s error=%s", assignment.JobID, err.Error()) return err } log.Printf("RUN phase=job status=execute_done job=%s state=%s errorCode=%s message=%s cancelledByPlatform=%t", assignment.JobID, execution.State, safeOptional(execution.ErrorCode), safeOptional(execution.Message), cancelledByPlatform) @@ -559,13 +559,13 @@ func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.Run resultRequest := LifecycleResultRequest(assignment, state.SessionToken, execution) log.Printf("RUN phase=job status=result_store job=%s state=%s", assignment.JobID, resultRequest.State) if err := worker.journal.StorePendingResult(resultRequest, execution.ActivationManifest); err != nil { - log.Printf("RUN phase=job status=result_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=result_store_failed job=%s error=%s", assignment.JobID, err.Error()) return err } log.Printf("RUN phase=job status=result_submit job=%s state=%s", assignment.JobID, resultRequest.State) result, err := worker.client.CompleteJob(ctx, resultRequest) if err != nil { - log.Printf("RUN phase=job status=result_submit_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=result_submit_failed job=%s error=%s", assignment.JobID, err.Error()) return err } if !result.Accepted { @@ -574,14 +574,14 @@ func (worker *Worker) runAssignment(ctx context.Context, assignment protocol.Run } log.Printf("RUN phase=job status=result_accepted job=%s state=%s", assignment.JobID, resultRequest.State) if err := worker.journal.Delete(assignment.JobID); err != nil { - log.Printf("RUN phase=job status=journal_delete_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job status=journal_delete_failed job=%s error=%s", assignment.JobID, err.Error()) return err } log.Printf("RUN phase=job status=complete job=%s state=%s", assignment.JobID, resultRequest.State) if execution.ActivationManifest != "" { log.Printf("RUN phase=self_update status=activate_start job=%s", assignment.JobID) if err := worker.executor.selfUpdateActivator.Activate(execution.ActivationManifest); err != nil { - log.Printf("RUN phase=self_update status=activate_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=self_update status=activate_failed job=%s error=%s", assignment.JobID, err.Error()) return fmt.Errorf("launch self-update helper: %w", err) } worker.restartMu.Lock() @@ -598,7 +598,7 @@ func (worker *Worker) executeWithJobPolling(ctx context.Context, assignment prot pollCancel := func() { state, err := worker.registeredState() if err != nil { - log.Printf("RUN phase=job.cancel_poll status=skipped_unregistered job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job.cancel_poll status=skipped_unregistered job=%s error=%s", assignment.JobID, err.Error()) return } log.Printf("RUN phase=job.cancel_poll status=starting job=%s", assignment.JobID) @@ -610,7 +610,7 @@ func (worker *Worker) executeWithJobPolling(ctx context.Context, assignment prot Attempt: assignment.Attempt, }) if err != nil { - log.Printf("RUN phase=job.cancel_poll status=failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job.cancel_poll status=failed job=%s error=%s", assignment.JobID, err.Error()) return } if err == nil && response.HasCancel { @@ -637,7 +637,7 @@ func (worker *Worker) executeWithJobPolling(ctx context.Context, assignment prot log.Printf("RUN phase=job.execute status=worker_finished job=%s state=%s", assignment.JobID, execution.State) return execution, assignment, cancelledByPlatform, nil case <-ctx.Done(): - log.Printf("RUN phase=job.execute status=context_done job=%s error=%s", assignment.JobID, RedactText(ctx.Err().Error())) + log.Printf("RUN phase=job.execute status=context_done job=%s error=%s", assignment.JobID, ctx.Err().Error()) cancel() execution := <-executionCh return execution, assignment, cancelledByPlatform, ctx.Err() @@ -652,7 +652,7 @@ func (worker *Worker) executeWithJobPolling(ctx context.Context, assignment prot state, err := worker.registeredState() if err != nil { cancel() - log.Printf("RUN phase=job.execute status=lease_renew_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job.execute status=lease_renew_failed job=%s error=%s", assignment.JobID, err.Error()) return LifecycleExecutionResult{}, assignment, cancelledByPlatform, err } progress, err := worker.client.UpdateJobProgress(ctx, protocol.RunJobProgressRequest{ @@ -669,14 +669,14 @@ func (worker *Worker) executeWithJobPolling(ctx context.Context, assignment prot if err == nil { err = fmt.Errorf("job lease renewal was not accepted") } - log.Printf("RUN phase=job.execute status=lease_renew_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job.execute status=lease_renew_failed job=%s error=%s", assignment.JobID, err.Error()) return LifecycleExecutionResult{}, assignment, cancelledByPlatform, err } assignment = progress.Job log.Printf("RUN phase=job.execute status=lease_renewed job=%s percent=%d", assignment.JobID, assignment.Progress.Percent) if err := worker.journal.Store(assignment); err != nil { cancel() - log.Printf("RUN phase=job.execute status=journal_store_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=job.execute status=journal_store_failed job=%s error=%s", assignment.JobID, err.Error()) return LifecycleExecutionResult{}, assignment, cancelledByPlatform, err } } @@ -880,7 +880,7 @@ func (worker *Worker) ReconcileOnce(ctx context.Context) error { ActiveJobs: worker.journal.ReconcileEntries(), }) if err != nil { - log.Printf("RUN phase=reconcile status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + log.Printf("RUN phase=reconcile status=failed endpoint=%s error=%s", state.RunEndpointID, err.Error()) return err } if !response.Accepted { @@ -928,7 +928,7 @@ func (worker *Worker) RecoverActiveJobs(ctx context.Context) error { pending.SessionToken = state.SessionToken result, err := worker.client.CompleteJob(ctx, pending) if err != nil { - log.Printf("RUN phase=recover status=result_submit_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=recover status=result_submit_failed job=%s error=%s", assignment.JobID, err.Error()) return err } if !result.Accepted { @@ -950,7 +950,7 @@ func (worker *Worker) RecoverActiveJobs(ctx context.Context) error { } log.Printf("RUN phase=recover status=rerun_active_job job=%s capability=%s", assignment.JobID, assignment.Capability) if err := worker.runAssignment(ctx, assignment); err != nil { - log.Printf("RUN phase=recover status=rerun_failed job=%s error=%s", assignment.JobID, RedactText(err.Error())) + log.Printf("RUN phase=recover status=rerun_failed job=%s error=%s", assignment.JobID, err.Error()) return err } } @@ -973,11 +973,11 @@ func (worker *Worker) Run(ctx context.Context) error { return err } if err := worker.reportAutonomousProcessObservations(ctx); err != nil { - log.Printf("RUN phase=autonomous_lifecycle.observation status=degraded error=%s", RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle.observation status=degraded error=%s", err.Error()) } worker.reportMetricsDegraded(ctx, "startup") if err := MarkSelfUpdateHealthy(worker.cfg.UpdateHealthFile); err != nil { - log.Printf("RUN phase=self_update_health status=mark_failed error=%s", RedactText(err.Error())) + log.Printf("RUN phase=self_update_health status=mark_failed error=%s", err.Error()) return err } if err := worker.reportRunUpdateHealth(ctx); err != nil { @@ -1004,7 +1004,7 @@ func (worker *Worker) Run(ctx context.Context) error { for { select { case <-ctx.Done(): - log.Printf("RUN phase=run status=context_done error=%s", RedactText(ctx.Err().Error())) + log.Printf("RUN phase=run status=context_done error=%s", ctx.Err().Error()) return ctx.Err() case err := <-jobDone: log.Printf("RUN phase=run status=job_loop_done error=%s", errorSummary(err)) @@ -1022,7 +1022,7 @@ func (worker *Worker) Run(ctx context.Context) error { continue } if err := worker.reportAutonomousProcessObservations(ctx); err != nil { - log.Printf("RUN phase=autonomous_lifecycle.observation status=degraded error=%s", RedactText(err.Error())) + log.Printf("RUN phase=autonomous_lifecycle.observation status=degraded error=%s", err.Error()) } worker.reportMetricsDegraded(ctx, "heartbeat") heartbeatTicker.Reset(heartbeatInterval) @@ -1054,7 +1054,7 @@ func (worker *Worker) reportRunUpdateHealth(ctx context.Context) error { Version: worker.cfg.Version, }) if err != nil { - log.Printf("RUN phase=self_update_health status=failed job=%s error=%s", worker.cfg.UpdateJobID, RedactText(err.Error())) + log.Printf("RUN phase=self_update_health status=failed job=%s error=%s", worker.cfg.UpdateJobID, err.Error()) return err } if !response.Accepted || response.JobID != worker.cfg.UpdateJobID { @@ -1088,19 +1088,19 @@ func (worker *Worker) runControlStreamLoop(ctx context.Context, wake chan<- stru return nil }) if ctx.Err() != nil { - log.Printf("RUN phase=control_stream status=context_done error=%s", RedactText(ctx.Err().Error())) + log.Printf("RUN phase=control_stream status=context_done error=%s", 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())) + log.Printf("RUN phase=control_stream status=session_invalid endpoint=%s error=%s", state.RunEndpointID, 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())) + log.Printf("RUN phase=control_stream status=reregister_failed endpoint=%s error=%s", state.RunEndpointID, refreshErr.Error()) } lastSeq = 0 signalControlWake(wake) } else { - log.Printf("RUN phase=control_stream status=failed endpoint=%s error=%s", state.RunEndpointID, RedactText(err.Error())) + log.Printf("RUN phase=control_stream status=failed endpoint=%s error=%s", state.RunEndpointID, err.Error()) } } if waitErr := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); waitErr != nil { @@ -1121,21 +1121,21 @@ func (worker *Worker) runJobLoop(ctx context.Context, interval time.Duration, wa for { select { case <-ctx.Done(): - log.Printf("RUN phase=job_loop status=context_done error=%s", RedactText(ctx.Err().Error())) + log.Printf("RUN phase=job_loop status=context_done error=%s", ctx.Err().Error()) return ctx.Err() default: } if worker.journal.ActiveCount() > 0 { log.Printf("RUN phase=job_loop status=active_jobs activeJobs=%d", worker.journal.ActiveCount()) if err := worker.ReconcileOnce(ctx); err != nil { - log.Printf("RUN phase=job_loop status=reconcile_failed error=%s", RedactText(err.Error())) + log.Printf("RUN phase=job_loop status=reconcile_failed error=%s", err.Error()) if err := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); err != nil { return err } continue } if err := worker.RecoverActiveJobs(ctx); err != nil { - log.Printf("RUN phase=job_loop status=recover_failed error=%s", RedactText(err.Error())) + log.Printf("RUN phase=job_loop status=recover_failed error=%s", err.Error()) if err := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); err != nil { return err } @@ -1145,7 +1145,7 @@ func (worker *Worker) runJobLoop(ctx context.Context, interval time.Duration, wa claimStartedAt := time.Now() handled, err := worker.claimAndRunOnce(ctx, 0) if err != nil { - log.Printf("RUN phase=job_loop status=claim_failed error=%s", RedactText(err.Error())) + log.Printf("RUN phase=job_loop status=claim_failed error=%s", err.Error()) if err := waitWorkerLoop(ctx, boundedRetryBackoff(worker.cfg.RetryBackoff)); err != nil { return err } @@ -1248,7 +1248,7 @@ func (client sessionLogBatchClient) IngestLogBatch(ctx context.Context, batch pr } response, err := client.client.IngestLogBatch(ctx, batch) if err != nil && logBatchNotFoundError(err) { - log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=platform_not_found error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, RedactText(err.Error())) + log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=platform_not_found error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, err.Error()) return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection("platform_not_found", err) } if err != nil && (logBatchSequenceGapError(err) || logBatchAcknowledgedRangeConflict(err)) { @@ -1256,15 +1256,15 @@ func (client sessionLogBatchClient) IngestLogBatch(ctx context.Context, batch pr if logBatchAcknowledgedRangeConflict(err) { reason = "platform_acknowledged_range_conflict" } - log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=%s error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, reason, RedactText(err.Error())) + log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=%s error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, reason, err.Error()) return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection(reason, err) } if err != nil && logBatchLegacySessionMetadataError(err) && strings.TrimSpace(batch.LogSessionID) == "" && !batch.SessionStartedAt.IsZero() { - log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=legacy_session_metadata error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, RedactText(err.Error())) + log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=legacy_session_metadata error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, err.Error()) return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection("legacy_session_metadata", err) } if err != nil && logBatchSessionMetadataMismatchError(err) { - log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=session_metadata_mismatch error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, RedactText(err.Error())) + log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=session_metadata_mismatch error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, err.Error()) return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection("session_metadata_mismatch", err) } return response, err @@ -1362,7 +1362,7 @@ func (worker *Worker) runDurableUploaders(ctx context.Context, done chan<- struc case <-ticker.C: state, err := worker.registeredState() if err != nil { - log.Printf("RUN phase=durable_uploaders status=skipped_unregistered error=%s", RedactText(err.Error())) + log.Printf("RUN phase=durable_uploaders status=skipped_unregistered error=%s", err.Error()) continue } if hasLogSink && hasLogClient { @@ -1372,7 +1372,7 @@ func (worker *Worker) runDurableUploaders(ctx context.Context, done chan<- struc client := sessionLogBatchClient{client: logClient, progressClient: progressClient, runEndpointID: state.RunEndpointID, sessionToken: state.SessionToken} flushed, err := logSink.Spool.Flush(flushCtx, client) if err != nil { - log.Printf("RUN phase=durable_uploaders.logs status=flush_failed flushed=%d durationMs=%d error=%s", flushed, time.Since(startedAt).Milliseconds(), RedactText(err.Error())) + log.Printf("RUN phase=durable_uploaders.logs status=flush_failed flushed=%d durationMs=%d error=%s", flushed, time.Since(startedAt).Milliseconds(), err.Error()) } else if flushed > 0 { log.Printf("RUN phase=durable_uploaders.logs status=flushed count=%d durationMs=%d", flushed, time.Since(startedAt).Milliseconds()) } @@ -1385,7 +1385,7 @@ func (worker *Worker) runDurableUploaders(ctx context.Context, done chan<- struc client := sessionArtifactChunkClient{client: artifactClient, runEndpointID: state.RunEndpointID, sessionToken: state.SessionToken} flushed, err := artifactHook.Queue.Flush(flushCtx, client) if err != nil { - log.Printf("RUN phase=durable_uploaders.artifacts status=flush_failed flushed=%d durationMs=%d error=%s", flushed, time.Since(startedAt).Milliseconds(), RedactText(err.Error())) + log.Printf("RUN phase=durable_uploaders.artifacts status=flush_failed flushed=%d durationMs=%d error=%s", flushed, time.Since(startedAt).Milliseconds(), err.Error()) } else if flushed > 0 { log.Printf("RUN phase=durable_uploaders.artifacts status=flushed count=%d durationMs=%d", flushed, time.Since(startedAt).Milliseconds()) } @@ -1542,7 +1542,7 @@ func (sink *LiveLogSink) dispatch() { _, err := sink.Client.RelayLiveLogBatch(ctx, batch) cancel() if err != nil { - log.Printf("RUN phase=live_log_relay status=dropped stream=%s sequence=%d error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, RedactText(err.Error())) + log.Printf("RUN phase=live_log_relay status=dropped stream=%s sequence=%d error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, err.Error()) } } } diff --git a/runtime/workspace_seed.go b/runtime/workspace_seed.go index 4133446..3b7170a 100644 --- a/runtime/workspace_seed.go +++ b/runtime/workspace_seed.go @@ -34,12 +34,12 @@ func MaterializeWorkspaceSeed(cfg config.Config) error { log.Printf("RUN phase=workspace_seed status=decoding workspace=%s encodedBytes=%d componentKey=%s", safeOptional(cfg.WorkspaceRoot), len(encoded), safeOptional(cfg.ComponentKey)) payload, err := base64.StdEncoding.DecodeString(encoded) if err != nil { - log.Printf("RUN phase=workspace_seed status=decode_failed workspace=%s error=%s", safeOptional(cfg.WorkspaceRoot), RedactText(err.Error())) + log.Printf("RUN phase=workspace_seed status=decode_failed workspace=%s error=%s", safeOptional(cfg.WorkspaceRoot), err.Error()) return fmt.Errorf("decode workspace seed: %w", err) } var files []workspaceSeedFile if err := json.Unmarshal(payload, &files); err != nil { - log.Printf("RUN phase=workspace_seed status=manifest_failed workspace=%s payloadBytes=%d error=%s", safeOptional(cfg.WorkspaceRoot), len(payload), RedactText(err.Error())) + log.Printf("RUN phase=workspace_seed status=manifest_failed workspace=%s payloadBytes=%d error=%s", safeOptional(cfg.WorkspaceRoot), len(payload), err.Error()) return fmt.Errorf("decode workspace seed manifest: %w", err) } log.Printf("RUN phase=workspace_seed status=decoded workspace=%s payloadBytes=%d files=%d", safeOptional(cfg.WorkspaceRoot), len(payload), len(files)) @@ -53,7 +53,7 @@ func MaterializeWorkspaceSeed(cfg config.Config) error { } scope, err := seededWorkspaceScopeForFiles(cfg, files) if err != nil { - log.Printf("RUN phase=workspace_seed status=scope_failed workspace=%s server=%s componentKey=%s error=%s", safeOptional(cfg.WorkspaceRoot), safeOptional(cfg.ServerInstanceID), safeOptional(cfg.ComponentKey), RedactText(err.Error())) + log.Printf("RUN phase=workspace_seed status=scope_failed workspace=%s server=%s componentKey=%s error=%s", safeOptional(cfg.WorkspaceRoot), safeOptional(cfg.ServerInstanceID), safeOptional(cfg.ComponentKey), err.Error()) return err } log.Printf("RUN phase=workspace_seed status=scope_ready workspace=%s scope=%s files=%d", safeOptional(cfg.WorkspaceRoot), safeOptional(scope), len(files)) @@ -61,7 +61,7 @@ func MaterializeWorkspaceSeed(cfg config.Config) error { for index, file := range files { written, err := writeWorkspaceSeedFile(scope, file, index+1, len(files)) if err != nil { - log.Printf("RUN phase=workspace_seed.file status=failed index=%d total=%d path=%s error=%s", index+1, len(files), safeOptional(file.Path), RedactText(err.Error())) + log.Printf("RUN phase=workspace_seed.file status=failed index=%d total=%d path=%s error=%s", index+1, len(files), safeOptional(file.Path), err.Error()) return err } totalBytes += written diff --git a/spool/artifact_queue.go b/spool/artifact_queue.go index af04b7a..89a04c4 100644 --- a/spool/artifact_queue.go +++ b/spool/artifact_queue.go @@ -1,6 +1,7 @@ package spool import ( + "bytes" "context" "encoding/json" "fmt" @@ -37,28 +38,20 @@ func (queue ArtifactQueue) Enqueue(chunk protocol.ArtifactChunkUploadRequest) er } else if !os.IsNotExist(err) { return err } - tmp := path + ".tmp" - file, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) - if err != nil { - return fmt.Errorf("open artifact queue chunk: %w", err) - } - encodeErr := json.NewEncoder(file).Encode(chunk) - closeErr := file.Close() - if encodeErr != nil { - _ = os.Remove(tmp) - return fmt.Errorf("encode artifact queue chunk: %w", encodeErr) - } - if closeErr != nil { - _ = os.Remove(tmp) - return fmt.Errorf("close artifact queue chunk: %w", closeErr) - } - if err := syncFile(tmp); err != nil { - _ = os.Remove(tmp) + payloadPath := artifactChunkPayloadPath(path) + if err := writeArtifactQueueFile(payloadPath, chunk.Payload); err != nil { return err } - if err := os.Rename(tmp, path); err != nil { - _ = os.Remove(tmp) - return fmt.Errorf("commit artifact queue chunk: %w", err) + metadata := chunk + metadata.Payload = nil + var body bytes.Buffer + if err := json.NewEncoder(&body).Encode(metadata); err != nil { + _ = os.Remove(payloadPath) + return fmt.Errorf("encode artifact queue chunk metadata: %w", err) + } + if err := writeArtifactQueueFile(path, body.Bytes()); err != nil { + _ = os.Remove(payloadPath) + return err } return nil } @@ -105,6 +98,11 @@ func readArtifactChunk(path string) (protocol.ArtifactChunkUploadRequest, error) if err := json.NewDecoder(file).Decode(&chunk); err != nil { return protocol.ArtifactChunkUploadRequest{}, fmt.Errorf("decode artifact queue chunk: %w", err) } + payload, err := os.ReadFile(artifactChunkPayloadPath(path)) + if err != nil { + return protocol.ArtifactChunkUploadRequest{}, fmt.Errorf("read artifact queue chunk payload: %w", err) + } + chunk.Payload = payload return chunk, nil } @@ -123,18 +121,9 @@ func (queue ArtifactQueue) Pending() ([]protocol.ArtifactChunkUploadRequest, err sort.Strings(paths) chunks := make([]protocol.ArtifactChunkUploadRequest, 0, len(paths)) for _, path := range paths { - file, err := os.Open(path) + chunk, err := readArtifactChunk(path) if err != nil { - return nil, fmt.Errorf("open artifact queue chunk: %w", err) - } - var chunk protocol.ArtifactChunkUploadRequest - decodeErr := json.NewDecoder(file).Decode(&chunk) - closeErr := file.Close() - if decodeErr != nil { - return nil, fmt.Errorf("decode artifact queue chunk: %w", decodeErr) - } - if closeErr != nil { - return nil, fmt.Errorf("close artifact queue chunk: %w", closeErr) + return nil, err } chunks = append(chunks, chunk) } @@ -154,23 +143,17 @@ func (queue ArtifactQueue) Ack(response protocol.ArtifactChunkUploadResponse) er continue } path := filepath.Join(queue.dir, entry.Name()) - file, err := os.Open(path) + chunk, err := readArtifactChunk(path) if err != nil { - return fmt.Errorf("open artifact queue chunk: %w", err) - } - var chunk protocol.ArtifactChunkUploadRequest - decodeErr := json.NewDecoder(file).Decode(&chunk) - closeErr := file.Close() - if decodeErr != nil { - return fmt.Errorf("decode artifact queue chunk: %w", decodeErr) - } - if closeErr != nil { - return fmt.Errorf("close artifact queue chunk: %w", closeErr) + return err } if chunk.TransferID == response.TransferID && chunk.ArtifactID == response.ArtifactID && chunk.ChunkIndex == response.ChunkIndex { if err := os.Remove(path); err != nil { return fmt.Errorf("remove acknowledged artifact queue chunk: %w", err) } + if err := os.Remove(artifactChunkPayloadPath(path)); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove acknowledged artifact queue payload: %w", err) + } } } return nil @@ -181,3 +164,34 @@ func (queue ArtifactQueue) chunkPath(chunk protocol.ArtifactChunkUploadRequest) artifactID := sanitizeSegmentName(chunk.ArtifactID) return filepath.Join(queue.dir, fmt.Sprintf("%s-%s-%020d.json", transferID, artifactID, chunk.ChunkIndex)) } + +func artifactChunkPayloadPath(metadataPath string) string { + return strings.TrimSuffix(metadataPath, ".json") + ".bin" +} + +func writeArtifactQueueFile(path string, payload []byte) error { + tmp := path + ".tmp" + file, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) + if err != nil { + return fmt.Errorf("open artifact queue file: %w", err) + } + _, writeErr := file.Write(payload) + closeErr := file.Close() + if writeErr != nil { + _ = os.Remove(tmp) + return fmt.Errorf("write artifact queue file: %w", writeErr) + } + if closeErr != nil { + _ = os.Remove(tmp) + return fmt.Errorf("close artifact queue file: %w", closeErr) + } + if err := syncFile(tmp); err != nil { + _ = os.Remove(tmp) + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + return fmt.Errorf("commit artifact queue file: %w", err) + } + return nil +} diff --git a/spool/artifact_queue_test.go b/spool/artifact_queue_test.go index 45dae9b..58136db 100644 --- a/spool/artifact_queue_test.go +++ b/spool/artifact_queue_test.go @@ -1,6 +1,9 @@ package spool import ( + "bytes" + "os" + "strings" "testing" "browser.local/run/protocol" @@ -16,6 +19,17 @@ func TestArtifactQueueRetainsPendingAndRemovesAcknowledgedChunk(t *testing.T) { if err := queue.Enqueue(first); err != nil { t.Fatalf("enqueue first: %v", err) } + metadata, err := os.ReadFile(queue.chunkPath(first)) + if err != nil { + t.Fatalf("read queued metadata: %v", err) + } + if strings.Contains(string(metadata), "payload") { + t.Fatalf("queued artifact metadata must not JSON encode payload: %s", string(metadata)) + } + storedPayload, err := os.ReadFile(artifactChunkPayloadPath(queue.chunkPath(first))) + if err != nil || !bytes.Equal(storedPayload, first.Payload) { + t.Fatalf("queued raw payload mismatch payload=%q err=%v", storedPayload, err) + } if err := queue.Enqueue(second); err != nil { t.Fatalf("enqueue second: %v", err) }