diff --git a/platform/service/distribution_build_execution.go b/platform/service/distribution_build_execution.go index 0b2f8aa..b8f202d 100644 --- a/platform/service/distribution_build_execution.go +++ b/platform/service/distribution_build_execution.go @@ -88,7 +88,16 @@ func (svc *CoreService) executeDistributionBuild(job domain.Job) error { if isTerminalJobState(job.State) { return nil } - payload, buildErr := svc.configuredDistributionBuilder().Build(input) + builder := svc.configuredDistributionBuilder() + var payload []byte + var buildErr error + if progressBuilder, ok := builder.(distributionBuilderWithProgress); ok { + payload, buildErr = progressBuilder.BuildWithProgress(input, func(progress DistributionBuildProgress) { + _ = svc.updateDistributionBuildProgress(job, progress) + }) + } else { + payload, buildErr = builder.Build(input) + } if buildErr != nil { return svc.failDistributionBuildJob(job, builderJobFailureMessage(buildErr)) } @@ -432,7 +441,7 @@ func (svc *CoreService) markDistributionBuildRunning(job *domain.Job) error { } current.State = domain.JobStateRunning current.Attempt = maxInt(current.Attempt, 1) - current.Progress = domain.JobProgress{Percent: 5, Phase: current.Progress.Phase, Message: "platform builder started"} + current.Progress = domain.JobProgress{Percent: 5, Phase: current.Progress.Phase, Message: "env_check: platform builder started"} current.UpdatedAt = stamp if err := svc.updateScheduledJob(current); err != nil { return err @@ -444,6 +453,29 @@ func (svc *CoreService) markDistributionBuildRunning(job *domain.Job) error { return nil } +func (svc *CoreService) updateDistributionBuildProgress(job domain.Job, progress DistributionBuildProgress) error { + stamp := svc.now() + svc.jobMu.Lock() + defer svc.jobMu.Unlock() + + current, err := svc.store.Jobs().Get(job.ID) + if err != nil { + return err + } + if isTerminalJobState(current.State) { + return nil + } + if progress.Percent < current.Progress.Percent { + progress.Percent = current.Progress.Percent + } + current.Progress = domain.JobProgress{Percent: progress.Percent, Phase: current.Progress.Phase, Message: strings.TrimSpace(progress.Message)} + current.UpdatedAt = stamp + if err := svc.updateScheduledJob(current); err != nil { + return err + } + return svc.projectDistributionBuildProgress(current, stamp) +} + func (svc *CoreService) succeedDistributionBuildJob(job domain.Job, artifactID string) error { stamp := svc.now() svc.jobMu.Lock() diff --git a/platform/service/distribution_build_execution_test.go b/platform/service/distribution_build_execution_test.go index 997f172..4c5d5d3 100644 --- a/platform/service/distribution_build_execution_test.go +++ b/platform/service/distribution_build_execution_test.go @@ -19,6 +19,12 @@ type captureDistributionBuilder struct { payload []byte } +type progressDistributionBuilder struct { + started chan struct{} + release <-chan struct{} + payload []byte +} + func (builder captureDistributionBuilder) Readiness() (bool, string) { return true, "" } @@ -31,6 +37,27 @@ func (builder captureDistributionBuilder) Build(input domain.DistributionBuildIn return domain.CopyBytes(builder.payload), nil } +func (builder progressDistributionBuilder) Readiness() (bool, string) { + return true, "" +} + +func (builder progressDistributionBuilder) Build(input domain.DistributionBuildInput) ([]byte, error) { + return builder.BuildWithProgress(input, nil) +} + +func (builder progressDistributionBuilder) BuildWithProgress(_ domain.DistributionBuildInput, progress func(DistributionBuildProgress)) ([]byte, error) { + if progress != nil { + progress(DistributionBuildProgress{Percent: 72, Message: "build_compile: compiling run target executable"}) + } + if builder.started != nil { + builder.started <- struct{}{} + } + if builder.release != nil { + <-builder.release + } + return domain.CopyBytes(builder.payload), nil +} + func TestCoreServiceKeepsPlatformBuildKeyOffMachineJobChannel(t *testing.T) { svc, session, instance := newDistributionTestFixture(t) plugin, err := svc.store.GamePlugins().Get(instance.PluginID) @@ -319,6 +346,39 @@ func TestCoreServiceDoesNotDuplicateInFlightPlatformBuild(t *testing.T) { completeDistributionBuild(t, svc, first, nil) } +func TestCoreServiceProjectsPlatformBuilderProgressBeforeCompletion(t *testing.T) { + svc, session, instance := newDistributionTestFixture(t) + started := make(chan struct{}, 1) + release := make(chan struct{}) + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) }) + svc.ConfigureDistributionBuilder(progressDistributionBuilder{started: started, release: release, payload: []byte("progress-platform-build")}) + + distribution, err := svc.GenerateRunDistributionForSession(session, domain.RunDistributionGenerateRequest{ + ServerInstanceID: instance.ID, + TargetOS: "linux", + TargetArch: "amd64", + IdempotencyKey: "platform-builder-progress", + }) + if err != nil { + t.Fatalf("generate platform distribution: %v", err) + } + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("platform builder did not report progress") + } + job, err := svc.GetJob(distribution.BuildJobID) + if err != nil { + t.Fatalf("get progress job: %v", err) + } + if job.State != domain.JobStateRunning || job.Progress.Percent != 72 || job.Progress.Message != "build_compile: compiling run target executable" { + t.Fatalf("expected in-flight platform builder progress, got %+v", job) + } + releaseOnce.Do(func() { close(release) }) + completeDistributionBuild(t, svc, distribution, nil) +} + func TestCoreServiceDiscardsBuildCompletedAfterKeyReset(t *testing.T) { svc, session, instance := newDistributionTestFixture(t) inputs := make(chan domain.DistributionBuildInput, 1) diff --git a/platform/service/distribution_builder.go b/platform/service/distribution_builder.go index f2f74e7..178f3ea 100644 --- a/platform/service/distribution_builder.go +++ b/platform/service/distribution_builder.go @@ -3,6 +3,7 @@ package service import ( "archive/tar" "archive/zip" + "bufio" "bytes" "compress/gzip" "context" @@ -11,10 +12,13 @@ import ( "encoding/json" "errors" "fmt" + "io" "os" "os/exec" "path/filepath" + "strconv" "strings" + "sync" "time" "browser.local/platform/domain" @@ -33,6 +37,15 @@ type DistributionBuilder interface { Build(input domain.DistributionBuildInput) ([]byte, error) } +type DistributionBuildProgress struct { + Percent int + Message string +} + +type distributionBuilderWithProgress interface { + BuildWithProgress(input domain.DistributionBuildInput, progress func(DistributionBuildProgress)) ([]byte, error) +} + // DockerDistributionBuilderConfig configures a container-per-build builder. type DockerDistributionBuilderConfig struct { DockerBinary string @@ -42,6 +55,7 @@ type DockerDistributionBuilderConfig struct { Timeout time.Duration PlatformURL string CommandRunner func(ctx context.Context, name string, args ...string) ([]byte, error) + CommandStream func(ctx context.Context, name string, args []string, onLine func(string)) ([]byte, error) } // DockerDistributionBuilder runs each build in a container from a pinned image, @@ -52,6 +66,7 @@ type DockerDistributionBuilder struct { } func NewDockerDistributionBuilder(config DockerDistributionBuilderConfig) *DockerDistributionBuilder { + customRunner := config.CommandRunner != nil if strings.TrimSpace(config.DockerBinary) == "" { config.DockerBinary = "docker" } @@ -61,6 +76,9 @@ func NewDockerDistributionBuilder(config DockerDistributionBuilderConfig) *Docke if config.CommandRunner == nil { config.CommandRunner = runCommandCombined } + if config.CommandStream == nil && !customRunner { + config.CommandStream = runCommandStreamCombined + } return &DockerDistributionBuilder{config: config} } @@ -68,6 +86,46 @@ func runCommandCombined(ctx context.Context, name string, args ...string) ([]byt return exec.CommandContext(ctx, name, args...).CombinedOutput() } +func runCommandStreamCombined(ctx context.Context, name string, args []string, onLine func(string)) ([]byte, error) { + cmd := exec.CommandContext(ctx, name, args...) + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, err + } + stderr, err := cmd.StderrPipe() + if err != nil { + return nil, err + } + if err := cmd.Start(); err != nil { + return nil, err + } + + var mu sync.Mutex + var output bytes.Buffer + var wg sync.WaitGroup + capture := func(reader io.Reader) { + defer wg.Done() + scanner := bufio.NewScanner(reader) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + line := scanner.Text() + mu.Lock() + output.WriteString(line) + output.WriteByte('\n') + mu.Unlock() + if onLine != nil { + onLine(line) + } + } + } + wg.Add(2) + go capture(stdout) + go capture(stderr) + err = cmd.Wait() + wg.Wait() + return output.Bytes(), err +} + func (builder *DockerDistributionBuilder) Readiness() (bool, string) { if strings.TrimSpace(builder.config.Image) == "" { return false, "platform builder image is not configured" @@ -124,9 +182,14 @@ func pinnedBuilderImage(image string) bool { } func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildInput) ([]byte, error) { + return builder.BuildWithProgress(input, nil) +} + +func (builder *DockerDistributionBuilder) BuildWithProgress(input domain.DistributionBuildInput, progress func(DistributionBuildProgress)) ([]byte, error) { if ready, reason := builder.Readiness(); !ready { return nil, validationError(reason) } + reportBuilderProgress(progress, 8, "env_check: platform builder readiness verified") sourceDir, err := filepath.Abs(strings.TrimSpace(builder.config.SourceDir)) if err != nil { return nil, validationError("platform builder run source directory is invalid") @@ -150,6 +213,7 @@ func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildIn } } defer func() { _ = os.RemoveAll(jobDir) }() + reportBuilderProgress(progress, 14, "env_check: platform builder workspace prepared") // The auth key reaches the container through a per-job input file, never // through a job-channel response to a machine-side endpoint or a container @@ -193,12 +257,15 @@ func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildIn ctx, cancel := context.WithTimeout(context.Background(), builder.config.Timeout) defer cancel() args := builder.containerArgs(input, sourceDir, inputDir, buildDir, outputDir, outputName) - if output, err := builder.config.CommandRunner(ctx, builder.config.DockerBinary, args...); err != nil { + reportBuilderProgress(progress, 18, "git_sync: platform builder container starting") + output, err := builder.runBuildCommand(ctx, args, progress) + if err != nil { if errors.Is(ctx.Err(), context.DeadlineExceeded) { return nil, validationError("platform builder timed out while building the distribution") } return nil, validationError("platform builder failed: " + safeBuilderFailure(output, input.AuthKey, builder.config.SourceDir, sourceDir, jobDir)) } + reportBuilderProgress(progress, 92, "package_finalize: reading platform builder output") binary, err := os.ReadFile(filepath.Join(outputDir, outputName)) if err != nil { return nil, validationError("platform builder did not produce a distribution executable") @@ -216,14 +283,74 @@ func (builder *DockerDistributionBuilder) Build(input domain.DistributionBuildIn return packageClientManagerDistribution(input.PackageFormat, outputName, binary, configPayload) } +func (builder *DockerDistributionBuilder) runBuildCommand(ctx context.Context, args []string, progress func(DistributionBuildProgress)) ([]byte, error) { + if builder.config.CommandStream == nil { + return builder.config.CommandRunner(ctx, builder.config.DockerBinary, args...) + } + return builder.config.CommandStream(ctx, builder.config.DockerBinary, args, func(line string) { + parsed, ok := parseBuilderProgressLine(line) + if ok { + reportBuilderProgress(progress, parsed.Percent, parsed.Message) + } + }) +} + +const builderProgressMarker = "__platform_builder_progress__|" + +func parseBuilderProgressLine(line string) (DistributionBuildProgress, bool) { + line = strings.TrimSpace(line) + if !strings.HasPrefix(line, builderProgressMarker) { + return DistributionBuildProgress{}, false + } + rest := strings.TrimPrefix(line, builderProgressMarker) + parts := strings.SplitN(rest, "|", 2) + if len(parts) != 2 { + return DistributionBuildProgress{}, false + } + percent, err := strconv.Atoi(strings.TrimSpace(parts[0])) + if err != nil { + return DistributionBuildProgress{}, false + } + message := strings.TrimSpace(parts[1]) + if message == "" { + return DistributionBuildProgress{}, false + } + if percent < 1 { + percent = 1 + } + if percent > 99 { + percent = 99 + } + return DistributionBuildProgress{Percent: percent, Message: message}, true +} + +func reportBuilderProgress(progress func(DistributionBuildProgress), percent int, message string) { + if progress == nil { + return + } + if percent < 1 { + percent = 1 + } + if percent > 99 { + percent = 99 + } + progress(DistributionBuildProgress{Percent: percent, Message: message}) +} + const distributionBuildScript = `#!/bin/sh set -eu +progress() { + printf '__platform_builder_progress__|%s|%s\n' "$1" "$2" +} + auth_key="$(cat /workspace/input/auth-key)" if [ "$COMPONENT_KIND" = "run" ]; then + progress 24 'git_sync: preparing run source snapshot' rm -rf /workspace/build/run-source mkdir -p /workspace/build/run-source cp -R /workspace/source/. /workspace/build/run-source/ + progress 34 'env_check: injecting run build metadata' seed_b64="$(base64 /workspace/input/workspace-seed.json | tr -d '\n')" cat > /workspace/build/run-source/config/workspace_seed_generated.go <&2; exit 2 ;; esac cd /workspace/build +progress 24 'git_sync: fetching client-manager source' git init --quiet git remote add origin "$REPOSITORY_URL" git fetch --quiet --depth 1 origin "$SOURCE_REVISION" git checkout --quiet --detach FETCH_HEAD +progress 36 'env_check: writing client-manager configuration' { printf 'server_url: "%s"\n' "$PLATFORM_URL" printf 'server_instance_id: "%s"\n' "$SERVER_INSTANCE_ID" @@ -270,9 +402,12 @@ git checkout --quiet --detach FETCH_HEAD printf 'scum_client_machine_label: "managed-client"\n' printf 'ftp_provider: 3\n' } > config.yaml +progress 52 'deps_download: downloading Go modules' go mod download +progress 76 'build_compile: compiling client-manager executable' go build -trimpath -ldflags '-s -w' -o "/workspace/output/$OUTPUT_FILENAME" . cp config.yaml /workspace/output/config.yaml +progress 88 'package_finalize: client-manager package inputs written' ` func (builder *DockerDistributionBuilder) containerArgs(input domain.DistributionBuildInput, sourceDir string, inputDir string, buildDir string, outputDir string, outputName string) []string { @@ -391,7 +526,7 @@ func safeBuilderFailure(output []byte, sensitiveValues ...string) string { kept := make([]string, 0, len(lines)) for index := len(lines) - 1; index >= 0 && len(kept) < 3; index-- { line := strings.TrimSpace(lines[index]) - if line == "" || strings.Contains(line, "/workspace/input") || strings.Contains(line, "auth-key") { + if line == "" || strings.HasPrefix(line, builderProgressMarker) || strings.Contains(line, "/workspace/input") || strings.Contains(line, "auth-key") { continue } line = redactBuilderHostPaths(line) diff --git a/platform/service/distribution_builder_test.go b/platform/service/distribution_builder_test.go index 677f470..3c15552 100644 --- a/platform/service/distribution_builder_test.go +++ b/platform/service/distribution_builder_test.go @@ -185,6 +185,58 @@ func TestDockerDistributionBuilderKeepsSecretInIsolatedInput(t *testing.T) { } } +func TestDockerDistributionBuilderStreamsProgressMarkers(t *testing.T) { + sourceDir := createBuilderSource(t) + workspaceDir := t.TempDir() + var progress []DistributionBuildProgress + builder := NewDockerDistributionBuilder(DockerDistributionBuilderConfig{ + DockerBinary: "docker-test", + Image: "browser-platform-distribution-builder:1.0.0", + SourceDir: sourceDir, + WorkspaceDir: workspaceDir, + CommandRunner: func(_ context.Context, _ string, args ...string) ([]byte, error) { + if len(args) > 0 && (args[0] == "version" || args[0] == "image") { + return []byte("27.0.0"), nil + } + return nil, errors.New("unexpected non-stream command") + }, + CommandStream: func(_ context.Context, _ string, args []string, onLine func(string)) ([]byte, error) { + onLine("__platform_builder_progress__|72|build_compile: compiling run target executable") + outputDir := builderMountHostPath(t, args, "/workspace/output") + if err := os.WriteFile(filepath.Join(outputDir, "run"), []byte("compiled-run"), 0o700); err != nil { + return nil, err + } + return []byte("builder finished"), nil + }, + }) + payload, err := builder.BuildWithProgress(domain.DistributionBuildInput{ + JobID: "job-progress", + ComponentKind: domain.DistributionComponentRun, + PluginID: "game.scum", + TargetOS: "linux", + TargetArch: "amd64", + OutputFilename: "run", + AuthKey: "component-key", + }, func(item DistributionBuildProgress) { + progress = append(progress, item) + }) + if err != nil { + t.Fatalf("build with progress: %v", err) + } + if string(payload) != "compiled-run" { + t.Fatalf("unexpected built payload %q", payload) + } + found := false + for _, item := range progress { + if item.Percent == 72 && item.Message == "build_compile: compiling run target executable" { + found = true + } + } + if !found { + t.Fatalf("streamed progress marker was not reported: %+v", progress) + } +} + func TestDockerDistributionBuilderRedactsFailureAndTimeout(t *testing.T) { sourceDir := createBuilderSource(t) workspaceDir := t.TempDir() diff --git a/platform_web/components/RuntimeTaskProgress.tsx b/platform_web/components/RuntimeTaskProgress.tsx index 51f4270..08dabbb 100644 --- a/platform_web/components/RuntimeTaskProgress.tsx +++ b/platform_web/components/RuntimeTaskProgress.tsx @@ -291,7 +291,7 @@ export function useRuntimeTaskController() { status: "succeeded", percent: 100, stageStatus: Object.fromEntries(stages.map((item) => [item.key, "completed" as RuntimeTaskStageStatus])), - logs: appendRuntimeLog(current.logs, "构建产物已由 run worker 上传") + logs: appendRuntimeLog(current.logs, "构建产物已由 platform builder 登记") } : current );