From cc53e823bc4453bc8bca589b32e4d872e72e10f6 Mon Sep 17 00:00:00 2001 From: Victor Garcia Date: Thu, 29 Oct 2026 16:35:05 -0600 Subject: [PATCH 1/2] fix(worker): expose durable publish recovery and track command processes --- README.md | 29 ++ cmd/jig/crash_reconcile_test.go | 160 +++++++ cmd/jig/publish_retry.go | 223 ++++++++++ cmd/jig/publish_retry_test.go | 401 ++++++++++++++++++ cmd/jig/worker.go | 27 +- docs/dogfooding.md | 23 +- docs/quickstart.md | 21 + internal/controlplane/embedded.go | 2 +- internal/controlplane/publish_ledger.go | 25 +- .../controlplane/publish_retry_sweep_test.go | 282 ++++++++++++ internal/controlplane/store.go | 2 +- internal/controlplane/sweep.go | 62 ++- internal/engine/code_phase.go | 27 ++ internal/engine/command_record_test.go | 320 ++++++++++++++ internal/engine/gates.go | 30 +- internal/engine/phase.go | 50 ++- .../worker/publish_retry_recovery_test.go | 125 ++++++ migrations/006_publish_retry_marker.sql | 12 + 18 files changed, 1786 insertions(+), 35 deletions(-) create mode 100644 cmd/jig/crash_reconcile_test.go create mode 100644 cmd/jig/publish_retry.go create mode 100644 cmd/jig/publish_retry_test.go create mode 100644 internal/controlplane/publish_retry_sweep_test.go create mode 100644 internal/engine/command_record_test.go create mode 100644 internal/worker/publish_retry_recovery_test.go create mode 100644 migrations/006_publish_retry_marker.sql diff --git a/README.md b/README.md index 423e97a..2864c03 100644 --- a/README.md +++ b/README.md @@ -424,6 +424,35 @@ on branch `jig//`, and accepted work is published there: push, find-or-create the pull request by head ref, then verify the **remote** ref as proof. +An `accepted_unpublished` job — held, its pull request not opened, or CI +red — is finished by the publish-only retry, run as the worker that ran it: + +```sh +# stop the worker that ran the job (Ctrl-C), then, as that worker: +jig worker publish-retry --data ~/.jig/worker +``` + +Use the same `--data` directory and the same `--commit-author-name`, +`--commit-author-email`, and `--base-branch` that worker runs with. The +retry runs no agent and spends no inference — it needs no runtime CLI at +all — and re-leases the SAME attempt: same branch, same pull request, CI +judged on the branch's current head, never repaired or re-run. It exits 0 +when the job is `accepted`, 3 when publish still did not finish (the job +stays `accepted_unpublished`; retry again), and 1 when it could not run. +The data directory is locked while a worker or a retry holds it, so stop the +worker first. A retry that is interrupted mid-publish is returned to +`accepted_unpublished` by the control plane once its lease expires and the +sweeper notices (about a minute); retry again after that. There is no retry button in +the UI. + +One data directory is one worker: `jig worker` holds `/worker.lock` +for as long as it runs, and a second worker or a retry on the same +directory is refused. A worker that dies hard leaves its agents' and its +code and gate commands' process groups in the attempt manifests, and the +next start on that directory stops them before reconciling anything else — +as long as the group's leader is still alive; a background process whose +leader already exited cannot be verified as jig's and is left alone. + Two things a worker prints at startup deserve reading: orphan worktrees (reported, never deleted) and stray publish branches — attempt-scoped branches on a remote that no fenced push record explains, which is the diff --git a/cmd/jig/crash_reconcile_test.go b/cmd/jig/crash_reconcile_test.go new file mode 100644 index 0000000..ddc2584 --- /dev/null +++ b/cmd/jig/crash_reconcile_test.go @@ -0,0 +1,160 @@ +// crash_reconcile_test.go — a code phase's command outlives a hard-killed +// worker only until the worker restarts. The worker runs as a real child +// process (this test binary re-executed, TestHelperProcess), its code phase +// is a shell loop writing a heartbeat file, and the worker is SIGKILLed while +// that shell is alive: no drain, no cleanup, nothing but the manifest left +// behind. A worker restarted on the SAME data directory must find the group +// in the manifest, stop it, and the heartbeat must stop growing. +package main + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/StructuPath/jig/internal/controlplane" + "github.com/StructuPath/jig/internal/protocol" +) + +func TestARestartedWorkerStopsTheCodePhaseGroupAHardKilledWorkerLeftRunning(t *testing.T) { + repo := initRepo(t) + serverData, workerData, scratch := t.TempDir(), t.TempDir(), t.TempDir() + scriptRuntime(t) + base, stopServe := startServe(t, serverData, "--no-github-poll") + defer stopServe() + + pidFile, heartbeat := filepath.Join(scratch, "pid"), filepath.Join(scratch, "heartbeat") + source := "name: crash-fixture\nphases:\n - name: loop\n kind: code\n command: |\n" + + " echo $$ > '" + pidFile + "'; while :; do echo beat >> '" + heartbeat + "'; sleep 0.05; done\n" + + "acceptance: [all_phases_passed]\n" + var definition protocol.Definition + if err := apiCall(context.Background(), base, "POST", "/api/definitions", + controlplane.DefinitionInput{Source: source}, &definition); err != nil { + t.Fatal(err) + } + if err := apiCall(context.Background(), base, "POST", "/api/runs", controlplane.RunInvocation{ + DefinitionID: definition.ID, Targets: []controlplane.InvocationTarget{{Repository: repo}}, + }, &controlplane.RunView{}); err != nil { + t.Fatal(err) + } + + child, _ := startHelper(t, "jig", "JIG_TEST_HELPER_ARGS="+strings.Join( + []string{"worker", "--server", base, "--data", workerData}, "\x1f")) + group := awaitCommandGroup(t, pidFile) + // Defensive cleanup of THIS test's group only: the id came from the + // shell this test's definition started, and it is never 0, 1, or ours. + t.Cleanup(func() { + if group > 1 && group != int64(syscall.Getpgrp()) { + _ = syscall.Kill(-int(group), syscall.SIGKILL) + } + }) + awaitManifestGroup(t, workerData, group) + + if err := child.Process.Signal(syscall.SIGKILL); err != nil { + t.Fatal(err) + } + _ = child.Wait() + // The worker is gone; its command is not — that is the orphan. + if !growing(t, heartbeat) { + t.Fatal("the code phase's shell died with the worker; the fixture proves nothing") + } + + ctx, interrupt := context.WithCancel(context.Background()) + stdout, stderr := &syncBuffer{}, &syncBuffer{} + exited := make(chan int, 1) + go func() { + exited <- workerCommand(ctx, []string{"--server", base, "--data", workerData}, stdout, stderr) + }() + defer func() { + interrupt() + select { + case <-exited: + case <-time.After(60 * time.Second): + t.Error("the restarted worker did not stop") + } + }() + want := "stopped orphaned agent process group: " + strconv.FormatInt(group, 10) + deadline := time.Now().Add(30 * time.Second) + for !strings.Contains(stdout.String(), want) { + if time.Now().After(deadline) { + t.Fatalf("the restarted worker never reported %q\nstdout:\n%s\nstderr:\n%s", want, stdout, stderr) + } + select { + case code := <-exited: + t.Fatalf("the restarted worker exited %d\nstderr:\n%s", code, stderr) + case <-time.After(20 * time.Millisecond): + } + } + if err := syscall.Kill(-int(group), 0); err != syscall.ESRCH { + t.Fatalf("process group %d still exists after reconciliation (kill 0: %v)", group, err) + } + if growing(t, heartbeat) { + t.Fatal("the heartbeat is still growing after reconciliation") + } +} + +func awaitCommandGroup(t *testing.T, pidFile string) int64 { + t.Helper() + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + if body, err := os.ReadFile(pidFile); err == nil && strings.HasSuffix(string(body), "\n") { + group, err := strconv.ParseInt(strings.TrimSpace(string(body)), 10, 64) + if err != nil { + t.Fatal(err) + } + return group + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("the code phase never started") + return 0 +} + +// awaitManifestGroup waits until an attempt manifest records the group live. +func awaitManifestGroup(t *testing.T, workerData string, group int64) { + t.Helper() + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + paths, _ := filepath.Glob(filepath.Join(workerData, "attempts", "*.json")) + for _, path := range paths { + var manifest struct { + ProcessGroups []int64 `json:"process_groups"` + ProcessGroupID int64 `json:"process_group_id"` + ProcessActive bool `json:"process_active"` + } + body, err := os.ReadFile(path) + if err != nil || json.Unmarshal(body, &manifest) != nil { + continue + } + if (manifest.ProcessActive && manifest.ProcessGroupID == group) || + slices.Contains(manifest.ProcessGroups, group) { + return + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("no attempt manifest recorded process group %d", group) +} + +// growing reports whether the file grows over a window several times the +// loop's period. +func growing(t *testing.T, path string) bool { + t.Helper() + size := func() int64 { + info, err := os.Stat(path) + if err != nil { + return 0 + } + return info.Size() + } + before := size() + time.Sleep(500 * time.Millisecond) + return size() > before +} diff --git a/cmd/jig/publish_retry.go b/cmd/jig/publish_retry.go new file mode 100644 index 0000000..83921fa --- /dev/null +++ b/cmd/jig/publish_retry.go @@ -0,0 +1,223 @@ +// publish_retry.go — `jig worker publish-retry` (R14): the operator's +// publish-only retry, run as the worker that owns the attempt, plus the +// data-directory lock it shares with `jig worker`. +// +// The retry is a one-shot over the worker's persisted identity and data +// directory. It deliberately skips everything `jig worker` does to be able to +// RUN attempts — runtime selection, the probe, registration, startup +// reconciliation — because a publish-only retry runs no phase: it needs no +// agent CLI and no agent auth, and registering would overwrite the running +// worker's advertised capabilities with an empty set. What it does need is +// the attempt's retained worktree, manifest, and repository cache, which +// only the owning worker's data directory holds — and which nothing else may +// be writing at the same time. That is what the lock is for. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "syscall" + + "github.com/StructuPath/jig/internal/protocol" + "github.com/StructuPath/jig/internal/worker" +) + +const workerPublishRetryUsage = `Usage: jig worker publish-retry [flags] + +Finish publishing an accepted_unpublished job — a held release, a pull +request that failed to open, CI that was red — without re-running any +phase. The SAME attempt is re-leased and only the publish steps the ledger +does not already prove are run: the same branch, the same pull request, CI +judged on the branch's current head. No agent starts and no inference is +spent; CI repair rounds and CI re-runs never happen on a retry. + +Run it as the worker that ran the attempt: the same --data directory, and +the same --commit-author-name/--commit-author-email/--base-branch that +worker uses. Stop that worker first — the data directory is locked while +either one runs. + +Flags: + --server control plane base URL (default http://127.0.0.1:8383) + --data the worker data directory the attempt ran in + (default ~/.jig/worker) + --commit-author-name author name on jig's publish commit + (default "jig") + --commit-author-email author email on jig's publish commit + (default "jig@localhost") + --base-branch pull-request base branch; empty resolves the + repository's default branch + +Exit codes: + 0 published: the attempt is accepted + 1 the retry could not run: the data directory is in use, the control + plane is unreachable, or it refused the retry (another worker's + attempt, a job that is not accepted_unpublished, an earlier retry + still holding its lease) + 2 usage error + 3 the retry ran but publish still did not finish; the job stays + accepted_unpublished and can be retried again +` + +// newPublishGateway builds the GitHub gateway publish runs through. A +// variable so the command tests can substitute a fake GitHub, exactly as the +// worker-wiring tests hand NewPublishingRunner one. +var newPublishGateway = func() worker.PullRequestGateway { return worker.NewGitHubCLIGateway() } + +func workerPublishRetryCommand(ctx context.Context, args []string, stdout, stderr io.Writer) int { + flags := flag.NewFlagSet("jig worker publish-retry", flag.ContinueOnError) + flags.SetOutput(stderr) + flags.Usage = func() { fmt.Fprint(stderr, workerPublishRetryUsage) } + serverURL := flags.String("server", "http://"+defaultServerAuthority, "control plane URL") + dataDir := flags.String("data", defaultWorkerDataDir(), "worker data directory") + authorName := flags.String("commit-author-name", "jig", "publish commit author name") + authorEmail := flags.String("commit-author-email", "jig@localhost", "publish commit author email") + baseBranch := flags.String("base-branch", "", "pull-request base branch") + if err := flags.Parse(args); err != nil { + return exitUsage + } + if flags.NArg() != 1 || strings.TrimSpace(flags.Arg(0)) == "" || *dataDir == "" { + flags.Usage() + return exitUsage + } + jobID := flags.Arg(0) + if ctx == nil { + ctx = context.Background() + } + + if err := os.MkdirAll(*dataDir, 0o700); err != nil { + fmt.Fprintf(stderr, "jig worker publish-retry: %v\n", err) + return exitInfraFailed + } + lock, err := lockWorkerData(*dataDir) + if err != nil { + fmt.Fprintf(stderr, "jig worker publish-retry: %v\n", err) + return exitInfraFailed + } + defer lock.release() + + host, err := worker.New(worker.Config{ + ServerURL: *serverURL, + DataDir: *dataDir, + WorkerVersion: version, + // No probe runs: a publish-only retry starts no agent. + Probes: []worker.RuntimeProbe{}, + Logger: commandLogger(stderr), + }) + if err != nil { + fmt.Fprintf(stderr, "jig worker publish-retry: %v\n", err) + return exitInfraFailed + } + // The retry runs to its recorded end on a context the interrupt does not + // cancel, as a worker's attempts do: a publish cut off after its push but + // before its record is exactly the state the retry exists to finish. A + // retry killed outright stays leased until the sweeper returns it to + // accepted_unpublished. + attempt, err := host.RetryPublish(context.WithoutCancel(ctx), jobID, newPublishGateway(), + worker.PublishOptions{ + CommitAuthorName: *authorName, + CommitAuthorEmail: *authorEmail, + BaseBranch: *baseBranch, + }) + if err != nil { + fmt.Fprintf(stderr, "jig worker publish-retry: %v\n", err) + var refused *worker.APIError + if errors.As(err, &refused) && refused.Code == "publish_retry_foreign_worker" { + fmt.Fprintf(stderr, "jig worker publish-retry: %s is worker %s; run the retry with the --data "+ + "directory of the worker that ran the attempt\n", *dataDir, host.ID()) + } + return exitInfraFailed + } + printPublishRetry(stdout, attempt) + if attempt.State != protocol.AttemptAccepted { + return exitRejected + } + return exitAccepted +} + +// printPublishRetry is the retry's answer: the attempt (the same one), its +// state, and where publish got to. +func printPublishRetry(stdout io.Writer, attempt protocol.Attempt) { + var document struct { + Publish worker.PublishSummary `json:"publish"` + } + _ = json.Unmarshal([]byte(attempt.Result), &document) + summary := document.Publish + fmt.Fprintf(stdout, "attempt: %s\n", attempt.ID) + fmt.Fprintf(stdout, "attempt_number: %d\n", attempt.AttemptNumber) + fmt.Fprintf(stdout, "state: %s\n", attempt.State) + fmt.Fprintf(stdout, "branch: %s\n", protocol.PublishBranch(attempt.JobID, attempt.AttemptNumber)) + if summary.PullRequestURL != "" { + fmt.Fprintf(stdout, "pull_request: %s\n", summary.PullRequestURL) + } + if summary.State != "" { + fmt.Fprintf(stdout, "publish: %s\n", summary.State) + } + if summary.Code != "" { + fmt.Fprintf(stdout, "publish_code: %s\n", summary.Code) + } + if summary.Detail != "" { + fmt.Fprintf(stdout, "publish_detail: %s\n", summary.Detail) + } +} + +func defaultWorkerDataDir() string { + if base := defaultDataDir(); base != "" { + return filepath.Join(base, "worker") + } + return "" +} + +// ---- the data-directory lock ----------------------------------------------- + +// workerDataLock is the exclusive flock on /worker.lock that `jig +// worker` and `jig worker publish-retry` hold for their whole lifetime. One +// data directory is one worker identity with one set of manifests, worktrees, +// and repository cache entries; two processes over it would race on all of +// them — a retry disposing the worktree a running worker's reconciliation is +// reading, or two publishers on one attempt branch. +// +// The lock lives on the open file description, so the kernel drops it +// however the process ends, SIGKILL included: there is no pid to check and +// no stale lock to clear. And it lives on the file, not the path: any alias +// of the directory — a symlink, a relative path — reaches the same inode and +// the same lock. +type workerDataLock struct{ file *os.File } + +func lockWorkerData(dataDir string) (*workerDataLock, error) { + path := filepath.Join(dataDir, "worker.lock") + file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, fmt.Errorf("open worker data lock: %w", err) + } + if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + file.Close() + if errors.Is(err, syscall.EWOULDBLOCK) { + return nil, fmt.Errorf("the worker data directory %s is in use by another jig worker process "+ + "(it holds %s); stop that `jig worker` or `jig worker publish-retry` first", dataDir, path) + } + return nil, fmt.Errorf("lock worker data directory %s: %w", dataDir, err) + } + // The pid is a note for an operator reading the file, never consulted. + if err := file.Truncate(0); err == nil { + _, _ = file.WriteString(strconv.Itoa(os.Getpid()) + "\n") + } + return &workerDataLock{file: file}, nil +} + +// release drops the lock. The file stays: removing it would let a racing +// opener lock an inode nobody else can find. +func (l *workerDataLock) release() { + if l == nil || l.file == nil { + return + } + _ = syscall.Flock(int(l.file.Fd()), syscall.LOCK_UN) + _ = l.file.Close() +} diff --git a/cmd/jig/publish_retry_test.go b/cmd/jig/publish_retry_test.go new file mode 100644 index 0000000..bc336d8 --- /dev/null +++ b/cmd/jig/publish_retry_test.go @@ -0,0 +1,401 @@ +// publish_retry_test.go — `jig worker publish-retry` end to end against a +// live `jig serve`, and the data-directory lock it shares with `jig worker`. +// Everything is production except the runtime (scripted, and removed before +// the retry) and GitHub (a fake gateway): a held release is published by the +// retry, red CI leaves it accepted_unpublished, and a second retry on green +// CI accepts it — the same attempt, branch, and pull request throughout, and +// not one more agent call. +package main + +import ( + "bufio" + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/StructuPath/jig/internal/controlplane" + "github.com/StructuPath/jig/internal/protocol" + "github.com/StructuPath/jig/internal/worker" +) + +const publishRetryDefinition = `name: publish-retry-e2e +roster: + builder: + model: test-model + system_prompt: Build. + user_prompt: "Build." + env: [PATH] + writes: ["jig-note.txt"] +phases: + - {name: build, kind: agent, owner: builder} + - name: classify + kind: code + reports_fields: true + command: | + echo '{"risk":"high"}' +acceptance: [all_phases_passed] +publish: + hold_when: "risk != low" + ci: + wait: true +` + +// retryGateway is the fake GitHub: one pull request per head, CI red until +// the test turns it green, and every judged head remembered. +type retryGateway struct { + mutex sync.Mutex + green bool + prs map[string]worker.PullRequest + created int + found int + judged []string +} + +func (g *retryGateway) FindOrCreatePullRequest(_ context.Context, request worker.PullRequestRequest) (worker.PullRequest, error) { + g.mutex.Lock() + defer g.mutex.Unlock() + if pr, exists := g.prs[request.Head]; exists { + g.found++ + return pr, nil + } + g.created++ + pr := worker.PullRequest{Number: g.created, URL: fmt.Sprintf("https://github.com/example/repo/pull/%d", g.created), State: "open"} + g.prs[request.Head] = pr + return pr, nil +} + +func (g *retryGateway) CommitChecks(_ context.Context, _ string, sha string) ([]worker.CICheck, error) { + g.mutex.Lock() + defer g.mutex.Unlock() + g.judged = append(g.judged, sha) + if g.green { + return []worker.CICheck{{Name: "test", Verdict: worker.CIPass, Conclusion: "success"}}, nil + } + return []worker.CICheck{{Name: "test", Verdict: worker.CIFail, Conclusion: "failure"}}, nil +} + +func (g *retryGateway) FailedCheckLogs(_ context.Context, _ string, checks []worker.CICheck) []worker.CICheck { + return checks +} + +func (g *retryGateway) ActionsJobRun(context.Context, string, int64) (int64, error) { + return 0, fmt.Errorf("unexpected workflow run lookup") +} + +func (g *retryGateway) RerunFailedJobs(context.Context, string, int64) error { + return fmt.Errorf("unexpected re-run request") +} + +func (g *retryGateway) counts() (created, found int, judged []string) { + g.mutex.Lock() + defer g.mutex.Unlock() + return g.created, g.found, append([]string(nil), g.judged...) +} + +// useGateway substitutes the fake GitHub for every command in this test. +func useGateway(t *testing.T, gateway worker.PullRequestGateway) { + t.Helper() + previous := newPublishGateway + newPublishGateway = func() worker.PullRequestGateway { return gateway } + t.Cleanup(func() { newPublishGateway = previous }) +} + +func publishRetry(t *testing.T, args ...string) (int, string, string) { + t.Helper() + stdout, stderr := &syncBuffer{}, &syncBuffer{} + code := workerCommand(context.Background(), append([]string{"publish-retry"}, args...), stdout, stderr) + t.Logf("jig worker publish-retry %v\nexit: %d\nstdout:\n%s\nstderr:\n%s", args, code, stdout, stderr) + return code, stdout.String(), stderr.String() +} + +// withoutAnyRuntime leaves PATH holding git alone and switches the scripted +// runtime off: a retry that tried to probe, seed, or start an agent would +// find nothing to run. +func withoutAnyRuntime(t *testing.T) { + t.Helper() + git, err := exec.LookPath("git") + if err != nil { + t.Fatal(err) + } + bin := t.TempDir() + if err := os.Symlink(git, filepath.Join(bin, "git")); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", bin) + t.Setenv(scriptedRuntimeEnv, "") +} + +func agentStarts(t *testing.T, base, attemptID string) int { + t.Helper() + starts := 0 + for _, event := range attemptEvents(t, base, attemptID) { + if event.Type == protocol.EventAgentStart { + starts++ + } + } + return starts +} + +func TestPublishRetryPublishesAHeldReleaseOnTheSameAttemptWithoutAnyRuntime(t *testing.T) { + repo := initRepo(t) + serverData, workerData := t.TempDir(), t.TempDir() + scriptRuntime(t, map[string]any{ + "files": map[string]any{"jig-note.txt": "held\n"}, + "text": envelopeJSON(t, map[string]any{"status": "success", "summary": "wrote the note"}), + }) + gateway := &retryGateway{prs: map[string]worker.PullRequest{}} + useGateway(t, gateway) + base, stopServe := startServe(t, serverData, "--no-github-poll") + defer stopServe() + + var definition protocol.Definition + if err := apiCall(context.Background(), base, "POST", "/api/definitions", + controlplane.DefinitionInput{Source: publishRetryDefinition}, &definition); err != nil { + t.Fatal(err) + } + var view controlplane.RunView + if err := apiCall(context.Background(), base, "POST", "/api/runs", controlplane.RunInvocation{ + DefinitionID: definition.ID, Instructions: "write the note", + Targets: []controlplane.InvocationTarget{{Repository: repo}}, + }, &view); err != nil { + t.Fatal(err) + } + jobID := view.Jobs[0].ID + + workerCtx, interrupt := context.WithCancel(context.Background()) + workerOut, workerErr := &syncBuffer{}, &syncBuffer{} + exited := make(chan int, 1) + go func() { + exited <- workerCommand(workerCtx, []string{"--server", base, "--data", workerData}, workerOut, workerErr) + }() + job := awaitTerminalJob(t, base, view.Run.ID, jobID) + if job.State != protocol.JobAcceptedUnpublished { + t.Fatalf("job = %q, want accepted_unpublished (held)\nworker stderr:\n%s", job.State, workerErr) + } + created, _, _ := gateway.counts() + branch := protocol.PublishBranch(jobID, 1) + if created != 0 || strings.Contains(gitIn(t, repo, "branch", "--list", branch), branch) { + t.Fatal("a held release was published before anyone retried it") + } + + // The retry refuses to run beside the worker that owns the directory — + // through an alias of the directory too. + alias := filepath.Join(t.TempDir(), "alias") + if err := os.Symlink(workerData, alias); err != nil { + t.Fatal(err) + } + for _, dir := range []string{workerData, alias} { + code, _, stderr := publishRetry(t, "--server", base, "--data", dir, jobID) + if code != exitInfraFailed || !strings.Contains(stderr, "in use by another jig worker") { + t.Fatalf("retry beside a running worker on %s: exit %d, want %d naming the lock", dir, code, exitInfraFailed) + } + } + interrupt() + select { + case code := <-exited: + if code != exitAccepted { + t.Fatalf("jig worker exited %d\nstderr:\n%s", code, workerErr) + } + case <-time.After(60 * time.Second): + t.Fatal("jig worker did not stop on interrupt") + } + + attemptID := latestAttempt(t, base, jobID).ID + startsBefore := agentStarts(t, base, attemptID) + if startsBefore != 1 { + t.Fatalf("the chain started %d agents, want 1", startsBefore) + } + withoutAnyRuntime(t) + + // Another worker's data directory is refused: only the owner has the + // attempt's retained state. + code, _, stderr := publishRetry(t, "--server", base, "--data", t.TempDir(), jobID) + if code != exitInfraFailed || !strings.Contains(stderr, "publish_retry_foreign_worker") || + !strings.Contains(stderr, "--data") { + t.Fatalf("foreign retry: exit %d, want %d naming the owning worker's --data", code, exitInfraFailed) + } + + // Retry 1 releases the hold: push, pull request, CI — red. + code, stdout, _ := publishRetry(t, "--server", base, "--data", workerData, + "--commit-author-name", "jig-test", "--commit-author-email", "jig@test", jobID) + if code != exitRejected || stdoutField(t, stdout, "attempt") != attemptID || + stdoutField(t, stdout, "state") != protocol.AttemptAcceptedUnpublished || + stdoutField(t, stdout, "publish_code") != "ci_failed" { + t.Fatalf("red-CI retry: exit %d, want %d with the same attempt left accepted_unpublished on ci_failed", + code, exitRejected) + } + pullRequest := stdoutField(t, stdout, "pull_request") + pushed := strings.TrimSpace(gitIn(t, repo, "rev-parse", "refs/heads/"+branch)) + + // Retry 2 on green CI accepts — the same pull request, judged on the + // branch's current head. + gateway.mutex.Lock() + gateway.green = true + gateway.mutex.Unlock() + code, stdout, _ = publishRetry(t, "--server", base, "--data", workerData, + "--commit-author-name", "jig-test", "--commit-author-email", "jig@test", jobID) + if code != exitAccepted || stdoutField(t, stdout, "attempt") != attemptID || + stdoutField(t, stdout, "state") != protocol.AttemptAccepted || + stdoutField(t, stdout, "pull_request") != pullRequest || + stdoutField(t, stdout, "branch") != branch { + t.Fatalf("green-CI retry: exit %d, want %d accepting the same attempt, branch, and pull request", + code, exitAccepted) + } + // The second retry re-enters on the ledger's proven pull_request step, + // so it neither creates nor even looks up a second one. + created, _, judged := gateway.counts() + if created != 1 { + t.Fatalf("pull requests created %d across the retries, want exactly 1", created) + } + if len(judged) == 0 || judged[len(judged)-1] != pushed { + t.Fatalf("CI judged %v, want the branch's current head %s last", judged, pushed) + } + if head := strings.TrimSpace(gitIn(t, repo, "rev-parse", "refs/heads/"+branch)); head != pushed { + t.Fatalf("the branch moved from %s to %s across the retries", pushed, head) + } + if starts := agentStarts(t, base, attemptID); starts != startsBefore { + t.Fatalf("agent starts went from %d to %d across publish-only retries", startsBefore, starts) + } + if attempt := latestAttempt(t, base, jobID); attempt.ID != attemptID || attempt.AttemptNumber != 1 { + t.Fatalf("the job's latest attempt is %s (#%d), want the original %s", attempt.ID, attempt.AttemptNumber, attemptID) + } + + // An accepted job has nothing left to publish. + code, _, stderr = publishRetry(t, "--server", base, "--data", workerData, jobID) + if code != exitInfraFailed || !strings.Contains(stderr, "publish_retry_not_allowed") { + t.Fatalf("retry of an accepted job: exit %d, want %d naming publish_retry_not_allowed", code, exitInfraFailed) + } +} + +func latestAttempt(t *testing.T, base, jobID string) protocol.Attempt { + t.Helper() + var detail controlplane.JobDetail + if err := apiCall(context.Background(), base, "GET", "/api/jobs/"+jobID, nil, &detail); err != nil { + t.Fatal(err) + } + attempts := detail.Attempts + if len(attempts) != 1 { + t.Fatalf("job %s has %d attempts, want exactly one: a publish-only retry never adds one", jobID, len(attempts)) + } + latest := attempts[0] + for _, attempt := range attempts[1:] { + if attempt.AttemptNumber > latest.AttemptNumber { + latest = attempt + } + } + return latest +} + +func TestPublishRetryRequiresExactlyOneJobID(t *testing.T) { + for _, args := range [][]string{{}, {"job-1", "job-2"}, {"--data", t.TempDir()}} { + if code, _, _ := publishRetry(t, args...); code != exitUsage { + t.Fatalf("publish-retry %v exited %d, want %d", args, code, exitUsage) + } + } +} + +// ---- the data-directory lock across processes ------------------------------- + +// helperEnv selects what TestHelperProcess does when this test binary is +// re-executed as a child: hold the data lock, or run jig. Unset, it skips. +const helperEnv = "JIG_TEST_HELPER" + +func TestHelperProcess(t *testing.T) { + switch os.Getenv(helperEnv) { + case "": + t.Skip("runs only as a child of another test") + case "hold-lock": + lock, err := lockWorkerData(os.Getenv("JIG_TEST_HELPER_DIR")) + if err != nil { + fmt.Println("error:", err) + os.Exit(1) + } + defer lock.release() + fmt.Println("locked") + select {} + case "jig": + os.Exit(run(context.Background(), strings.Split(os.Getenv("JIG_TEST_HELPER_ARGS"), "\x1f"))) + } +} + +// startHelper re-executes this test binary as a child in the given helper +// mode. The child is killed at cleanup if it is still running. +func startHelper(t *testing.T, mode string, env ...string) (*exec.Cmd, *bufio.Scanner) { + t.Helper() + command := exec.Command(os.Args[0], "-test.run=^TestHelperProcess$") + command.Env = append(append(os.Environ(), helperEnv+"="+mode), env...) + command.Stderr = os.Stderr + stdout, err := command.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := command.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = command.Process.Kill() + _ = command.Wait() + }) + return command, bufio.NewScanner(stdout) +} + +// Another PROCESS holding the lock refuses both a worker and a retry, by the +// directory's path or an alias of it; once that process is SIGKILLed, the +// kernel has dropped the lock and the directory is usable at once. +func TestTheWorkerDataLockHoldsAcrossProcessesAndDiesWithItsHolder(t *testing.T) { + dataDir := t.TempDir() + holder, lines := startHelper(t, "hold-lock", "JIG_TEST_HELPER_DIR="+dataDir) + if !lines.Scan() || lines.Text() != "locked" { + t.Fatalf("the holder did not take the lock: %q", lines.Text()) + } + alias := filepath.Join(t.TempDir(), "alias") + if err := os.Symlink(dataDir, alias); err != nil { + t.Fatal(err) + } + relative, err := filepath.Rel(mustGetwd(t), dataDir) + if err != nil { + t.Fatal(err) + } + for _, dir := range []string{dataDir, alias, relative} { + stdout, stderr := &syncBuffer{}, &syncBuffer{} + if code := workerCommand(context.Background(), []string{"--server", "http://127.0.0.1:1", "--data", dir}, + stdout, stderr); code != exitInfraFailed || !strings.Contains(stderr.String(), "in use by another jig worker") { + t.Fatalf("worker on %s beside the holder: exit %d stderr %q, want a lock refusal", dir, code, stderr) + } + if code, _, stderr := publishRetry(t, "--server", "http://127.0.0.1:1", "--data", dir, "job-1"); code != exitInfraFailed || + !strings.Contains(stderr, "in use by another jig worker") { + t.Fatalf("retry on %s beside the holder: exit %d, want a lock refusal", dir, code) + } + } + + if err := holder.Process.Signal(syscall.SIGKILL); err != nil { + t.Fatal(err) + } + _ = holder.Wait() + lock, err := lockWorkerData(alias) + if err != nil { + t.Fatalf("the lock outlived its SIGKILLed holder: %v", err) + } + lock.release() + // And a clean exit releases it the same way. + if lock, err = lockWorkerData(dataDir); err != nil { + t.Fatalf("a released lock could not be taken again: %v", err) + } + lock.release() +} + +func mustGetwd(t *testing.T) string { + t.Helper() + dir, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + return dir +} diff --git a/cmd/jig/worker.go b/cmd/jig/worker.go index 9441d75..3866531 100644 --- a/cmd/jig/worker.go +++ b/cmd/jig/worker.go @@ -41,10 +41,15 @@ import ( ) const workerUsage = `Usage: jig worker [--server ] [--data ] [--runtime ] [--capacity ] + jig worker publish-retry [flags] Run the worker host: register with the control plane, claim queued work, execute each attempt's phase chain in an isolated worktree, publish accepted -work, and reconcile retained worktrees at start. +work, and reconcile retained worktrees at start. The data directory is +locked while the worker runs: one directory, one worker process. + +publish-retry finishes an accepted_unpublished job's publish as this +worker, without re-running any phase; see jig worker publish-retry --help. Flags: --server control plane base URL (default http://127.0.0.1:8383) @@ -77,15 +82,14 @@ Exit codes: const workerDrainGrace = 2 * time.Minute func workerCommand(ctx context.Context, args []string, stdout, stderr io.Writer) int { + if len(args) > 0 && args[0] == "publish-retry" { + return workerPublishRetryCommand(ctx, args[1:], stdout, stderr) + } flags := flag.NewFlagSet("jig worker", flag.ContinueOnError) flags.SetOutput(stderr) flags.Usage = func() { fmt.Fprint(stderr, workerUsage) } serverURL := flags.String("server", "http://"+defaultServerAuthority, "control plane URL") - defaultWorkerData := "" - if base := defaultDataDir(); base != "" { - defaultWorkerData = filepath.Join(base, "worker") - } - dataDir := flags.String("data", defaultWorkerData, "worker data directory") + dataDir := flags.String("data", defaultWorkerDataDir(), "worker data directory") runtimeName := flags.String("runtime", runtimeClaudeCode, "agent CLI runtime") capacity := flags.Int("capacity", 1, "concurrent attempt slots") name := flags.String("name", "local", "worker name") @@ -109,6 +113,15 @@ func workerCommand(ctx context.Context, args []string, stdout, stderr io.Writer) fmt.Fprintf(stderr, "jig worker: %v\n", err) return exitInfraFailed } + // Held before the identity is read or anything is probed, until the + // worker returns: a second worker or a publish-only retry on the same + // directory would race this one's manifests and worktrees. + lock, err := lockWorkerData(*dataDir) + if err != nil { + fmt.Fprintf(stderr, "jig worker: %v\n", err) + return exitInfraFailed + } + defer lock.release() selected, err := selectRuntime(ctx, *runtimeName, !*noSeedAuth) if err != nil { fmt.Fprintf(stderr, "jig worker: %v\n", err) @@ -132,7 +145,7 @@ func workerCommand(ctx context.Context, args []string, stdout, stderr io.Writer) var host *worker.Worker engineRunner := workerAttemptRunner(ctx, *dataDir, selected, logger, func() *worker.Worker { return host }) - publisher := worker.NewPublishingRunner(engineRunner, worker.NewGitHubCLIGateway(), + publisher := worker.NewPublishingRunner(engineRunner, newPublishGateway(), worker.PublishOptions{ CommitAuthorName: *authorName, CommitAuthorEmail: *authorEmail, diff --git a/docs/dogfooding.md b/docs/dogfooding.md index b0c59da..d99978d 100644 --- a/docs/dogfooding.md +++ b/docs/dogfooding.md @@ -365,9 +365,26 @@ The last field is the one the report cannot make. jig does not watch CI or review after an accept, so a clean accept that you then had to rewrite counts as clean in the report. Record it here. -A publish-only retry is `POST /api/jobs//publish-retry`, or the -button in the UI. It judges CI on the branch's current head, and it never -repairs. +A publish-only retry runs on the worker that ran the job — the retained +worktree and the repository cache are in that worker's data directory: + +```sh +# stop the worker that ran the job (Ctrl-C), then, as that worker: +jig worker publish-retry --data ~/.jig/worker +``` + +Use the same `--data` directory and the same `--commit-author-name`, +`--commit-author-email`, and `--base-branch` that worker runs with. The +retry runs no agent and spends no inference — it needs no runtime CLI at +all — and re-leases the SAME attempt: same branch, same pull request, CI +judged on the branch's current head, never repaired or re-run. It exits 0 +when the job is `accepted`, 3 when publish still did not finish (the job +stays `accepted_unpublished`; retry again), and 1 when it could not run. +The data directory is locked while a worker or a retry holds it, so stop the +worker first. A retry that is interrupted mid-publish is returned to +`accepted_unpublished` by the control plane once its lease expires and the +sweeper notices (about a minute); retry again after that. There is no retry button in +the UI. --- diff --git a/docs/quickstart.md b/docs/quickstart.md index 50585a1..9763983 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -234,6 +234,27 @@ failing checks and every round in its result; push a fix to the `jig//` branch yourself and the publish retry re-checks CI on the new head (it never repairs). +The publish retry, for a held release or a red-CI job alike, runs as the +worker that ran the job: + +```sh +# stop the worker that ran the job (Ctrl-C), then, as that worker: +jig worker publish-retry --data ~/.jig/worker +``` + +Use the same `--data` directory and the same `--commit-author-name`, +`--commit-author-email`, and `--base-branch` that worker runs with. The +retry runs no agent and spends no inference — it needs no runtime CLI at +all — and re-leases the SAME attempt: same branch, same pull request, CI +judged on the branch's current head, never repaired or re-run. It exits 0 +when the job is `accepted`, 3 when publish still did not finish (the job +stays `accepted_unpublished`; retry again), and 1 when it could not run. +The data directory is locked while a worker or a retry holds it, so stop the +worker first. A retry that is interrupted mid-publish is returned to +`accepted_unpublished` by the control plane once its lease expires and the +sweeper notices (about a minute); retry again after that. There is no retry button in +the UI. + `Ctrl-C` stops the worker in order: no new claims, in-flight attempts cancelled, terminal states recorded, ephemeral scratch destroyed. diff --git a/internal/controlplane/embedded.go b/internal/controlplane/embedded.go index b38993c..5a9b9ac 100644 --- a/internal/controlplane/embedded.go +++ b/internal/controlplane/embedded.go @@ -686,7 +686,7 @@ func (s *Store) reclaimAbandonedDirectRuns(ctx context.Context, dataDir string) if err := removeDirectMarker(dataDir, value.jobID); err != nil { failures = append(failures, err) } - reclaimed = append(reclaimed, LostAttempt{AttemptID: value.attemptID, JobID: value.jobID}) + reclaimed = append(reclaimed, LostAttempt{AttemptID: value.attemptID, JobID: value.jobID, State: protocol.AttemptLost}) } return reclaimed, errors.Join(failures...) } diff --git a/internal/controlplane/publish_ledger.go b/internal/controlplane/publish_ledger.go index 2acb1b6..0317054 100644 --- a/internal/controlplane/publish_ledger.go +++ b/internal/controlplane/publish_ledger.go @@ -338,6 +338,11 @@ func (s *Store) publishRecords(ctx context.Context, where string, argument any) // // Re-leasing supersedes any previous token on that attempt: a stalled // publisher that wakes up afterwards is fenced out by digest mismatch. +// +// The re-leased attempt is marked publish_retry, so an interrupted retry is +// swept back to accepted_unpublished rather than lost (sweepAttempt): the +// marker, not the stored result, is what says "this lease only publishes", +// because an accepted_unpublished attempt may legitimately have no result. func (s *Store) RetryPublish( ctx context.Context, jobID string, input protocol.PublishRetryRequest, ) (protocol.PublishRetry, error) { @@ -363,6 +368,23 @@ func (s *Store) RetryPublish( return retry, unavailable(err) } if jobState != protocol.JobAcceptedUnpublished { + var inProgress int + if err := tx.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM attempts + WHERE job_id = ? AND state = 'running' AND publish_retry = 1 + `, jobID).Scan(&inProgress); err != nil { + return retry, unavailable(err) + } + if inProgress > 0 { + // An earlier retry still holds the lease — typically one whose + // process died mid-publish. Taking it over here would race a + // publisher that may still be alive; the sweeper returns it to + // accepted_unpublished once the lease has expired, and the retry + // is available again from there. + return retry, conflict("publish_retry_in_progress", + "a previous publish-only retry of this job still holds its lease; "+ + "retry after it expires and the sweeper returns the job to accepted_unpublished") + } return retry, conflict("publish_retry_not_allowed", "only an accepted_unpublished job has a publish-only retry") } @@ -392,7 +414,8 @@ func (s *Store) RetryPublish( expiry := now.Add(protocol.LeaseDuration).UnixMilli() result, err := tx.ExecContext(ctx, ` UPDATE attempts - SET state = 'running', lease_digest = ?, lease_expires_at = ?, completed_at = NULL, error = NULL + SET state = 'running', lease_digest = ?, lease_expires_at = ?, completed_at = NULL, error = NULL, + publish_retry = 1 WHERE id = ? AND state = 'accepted_unpublished' `, digestToken(input.LeaseToken), expiry, attemptID) if err != nil { diff --git a/internal/controlplane/publish_retry_sweep_test.go b/internal/controlplane/publish_retry_sweep_test.go new file mode 100644 index 0000000..687ce8d --- /dev/null +++ b/internal/controlplane/publish_retry_sweep_test.go @@ -0,0 +1,282 @@ +// publish_retry_sweep_test.go — an interrupted publish-only retry stays +// recoverable (R14): the sweeper returns its attempt to accepted_unpublished, +// same attempt and records, instead of `lost`, and the interrupted publisher +// is fenced out. The publish_retry marker — not the stored result — is the +// discriminator, because an accepted_unpublished attempt may have no result. +package controlplane + +import ( + "context" + "database/sql" + "sort" + "strings" + "testing" + + "github.com/StructuPath/jig/internal/protocol" + "github.com/StructuPath/jig/migrations" +) + +const tokenC = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" + +// sweepUntilSettled ticks the sweeper at heartbeat cadence until the attempt +// leaves `running`, and returns what the sweeper reported. +func sweepUntilSettled(t *testing.T, store *Store, clock *testClock, attemptID string) []LostAttempt { + t.Helper() + sweeper := NewSweeper(store) + maxTicks := int(protocol.LeaseDuration/protocol.HeartbeatInterval) + + protocol.MissedHeartbeatsBeforeSweep + 2 + var reported []LostAttempt + for i := 0; i < maxTicks; i++ { + clock.Advance(protocol.HeartbeatInterval) + swept, err := sweeper.Tick(context.Background()) + if err != nil { + t.Fatalf("sweep tick: %v", err) + } + reported = append(reported, swept...) + attempt, err := store.Attempt(context.Background(), attemptID) + if err != nil { + t.Fatalf("read attempt: %v", err) + } + if attempt.State != protocol.AttemptRunning { + return reported + } + } + t.Fatal("the sweeper never settled the attempt") + return nil +} + +func publishRetryMarker(t *testing.T, store *Store, attemptID string) int { + t.Helper() + var marker int + if err := store.db.QueryRow(`SELECT publish_retry FROM attempts WHERE id = ?`, attemptID).Scan(&marker); err != nil { + t.Fatalf("read publish_retry: %v", err) + } + return marker +} + +func storedResult(t *testing.T, store *Store, attemptID string) sql.NullString { + t.Helper() + var result sql.NullString + if err := store.db.QueryRow(`SELECT result FROM attempts WHERE id = ?`, attemptID).Scan(&result); err != nil { + t.Fatalf("read result: %v", err) + } + return result +} + +// An interrupted retry — from a previous result or from none at all — is +// swept back to accepted_unpublished on the same attempt, with its publish +// records, its result, and the run aggregate exactly as before the retry. The +// interrupted publisher can no longer heartbeat, record, or complete, and a +// second retry re-enters the same attempt where the first stopped. +func TestAnInterruptedPublishRetryIsSweptBackToAcceptedUnpublished(t *testing.T) { + for name, result := range map[string]string{ + "previous result": `{"changed_paths":["a.txt"]}`, + "no previous result": "", + } { + t.Run(name, func(t *testing.T) { + store, clock := newTestStore(t) + ctx := context.Background() + registerTestWorker(t, store, 1) + seedRun(t, store, "run-1", protocol.RunTarget{Repository: repoA, BaseSHA: shaA}) + claim := claimAndStart(t, store, "run-1", repoA) + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepPush, shaB, ""); err != nil { + t.Fatalf("record push: %v", err) + } + if _, err := store.CompleteAttempt(ctx, claim.Attempt.ID, protocol.CompleteAttemptRequest{ + LeaseToken: tokenA, State: protocol.AttemptAcceptedUnpublished, Result: result, + }); err != nil { + t.Fatalf("complete accepted_unpublished: %v", err) + } + before := storedResult(t, store, claim.Attempt.ID) + runBefore, err := store.Run(ctx, "run-1") + if err != nil { + t.Fatal(err) + } + + if _, err := store.RetryPublish(ctx, claim.Job.ID, + protocol.PublishRetryRequest{WorkerID: workerA, LeaseToken: tokenB}); err != nil { + t.Fatalf("publish retry: %v", err) + } + if marker := publishRetryMarker(t, store, claim.Attempt.ID); marker != 1 { + t.Fatalf("publish_retry = %d on the re-leased attempt, want 1", marker) + } + // While the interrupted retry still holds its lease, a second retry + // is refused with a reason an operator can act on. + if _, err := store.RetryPublish(ctx, claim.Job.ID, + protocol.PublishRetryRequest{WorkerID: workerA, LeaseToken: tokenC}); err == nil || + serviceCode(t, err) != "publish_retry_in_progress" { + t.Fatalf("retry over a live retry lease: err=%v, want publish_retry_in_progress", err) + } + + // The retry's process dies: no heartbeat, no completion. + swept := sweepUntilSettled(t, store, clock, claim.Attempt.ID) + if len(swept) != 1 || swept[0].AttemptID != claim.Attempt.ID || + swept[0].State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("sweeper reported %+v, want this attempt returned to accepted_unpublished", swept) + } + attempt, err := store.Attempt(ctx, claim.Attempt.ID) + if err != nil { + t.Fatal(err) + } + job, err := store.Job(ctx, claim.Job.ID) + if err != nil { + t.Fatal(err) + } + if attempt.State != protocol.AttemptAcceptedUnpublished || job.State != protocol.JobAcceptedUnpublished { + t.Fatalf("attempt=%q job=%q, want accepted_unpublished for both", attempt.State, job.State) + } + if !strings.Contains(attempt.Error, "publish retry interrupted") { + t.Fatalf("attempt error = %q, want the interruption named", attempt.Error) + } + if after := storedResult(t, store, claim.Attempt.ID); after != before { + t.Fatalf("result changed across the sweep: %+v -> %+v", before, after) + } + if marker := publishRetryMarker(t, store, claim.Attempt.ID); marker != 0 { + t.Fatalf("publish_retry = %d after the sweep, want 0", marker) + } + if runAfter, err := store.Run(ctx, "run-1"); err != nil || runAfter.State != runBefore.State { + t.Fatalf("run state %q -> %q (err %v), want it unchanged by the recovery", + runBefore.State, runAfter.State, err) + } + records, err := store.AttemptPublishRecords(ctx, claim.Attempt.ID) + if err != nil || len(records) != 1 || records[0].RemoteRef != shaB { + t.Fatalf("publish records = %+v (err %v), want the proven push kept", records, err) + } + + // The interrupted publisher wakes up: every fenced write refuses it. + if _, err := store.Heartbeat(ctx, claim.Attempt.ID, + protocol.HeartbeatRequest{LeaseToken: tokenB}); err == nil { + t.Fatal("a swept retry's heartbeat renewed its lease") + } + if _, err := recordStep(store, claim, tokenB, protocol.PublishStepPullRequest, "", + "https://github.com/example/repo-a/pull/7"); err == nil || + serviceCode(t, err) != "lease_not_owner" { + t.Fatalf("a swept retry recorded a step: err=%v, want lease_not_owner", err) + } + if _, err := store.CompleteAttempt(ctx, claim.Attempt.ID, protocol.CompleteAttemptRequest{ + LeaseToken: tokenB, State: protocol.AttemptAcceptedUnpublished, Result: "zombie", + }); err == nil || serviceCode(t, err) != "lease_not_owner" { + t.Fatalf("a swept retry completed: err=%v, want lease_not_owner", err) + } + + // The retry is available again, on the same attempt. + again, err := store.RetryPublish(ctx, claim.Job.ID, + protocol.PublishRetryRequest{WorkerID: workerA, LeaseToken: tokenC}) + if err != nil { + t.Fatalf("publish retry after recovery: %v", err) + } + if again.Attempt.ID != claim.Attempt.ID || len(again.Records) != 1 { + t.Fatalf("second retry = attempt %s records %+v, want the same attempt and its push", + again.Attempt.ID, again.Records) + } + // The superseded first token stays fenced out of the new lease. + if _, err := store.Heartbeat(ctx, claim.Attempt.ID, + protocol.HeartbeatRequest{LeaseToken: tokenB}); err == nil || + serviceCode(t, err) != "lease_not_owner" { + t.Fatalf("the first retry's token renewed the second's lease: err=%v", err) + } + if _, err := store.CompleteAttempt(ctx, claim.Attempt.ID, protocol.CompleteAttemptRequest{ + LeaseToken: tokenC, State: protocol.AttemptAcceptedUnpublished, Result: result, + }); err != nil { + t.Fatalf("complete the second retry: %v", err) + } + if marker := publishRetryMarker(t, store, claim.Attempt.ID); marker != 0 { + t.Fatalf("publish_retry = %d after completion, want 0", marker) + } + var attempts int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM attempts WHERE job_id = ?`, + claim.Job.ID).Scan(&attempts); err != nil || attempts != 1 { + t.Fatalf("the job holds %d attempts (err %v), want 1", attempts, err) + } + }) + } +} + +// Without the marker an expired running attempt is lost exactly as before — +// even one whose result column happens to be set, which is why the result is +// not the discriminator. An ordinary attempt that runs CI repair rounds is an +// ordinary running attempt and is covered by the same path. +func TestAnExpiredUnmarkedAttemptIsStillLost(t *testing.T) { + store, clock := newTestStore(t) + registerTestWorker(t, store, 1) + seedRun(t, store, "run-1", protocol.RunTarget{Repository: repoA, BaseSHA: shaA}) + claim := claimAndStart(t, store, "run-1", repoA) + if _, err := store.db.Exec(`UPDATE attempts SET result = '{"partial":true}' WHERE id = ?`, + claim.Attempt.ID); err != nil { + t.Fatal(err) + } + swept := sweepUntilSettled(t, store, clock, claim.Attempt.ID) + if len(swept) != 1 || swept[0].State != protocol.AttemptLost { + t.Fatalf("sweeper reported %+v, want the attempt lost", swept) + } + job, err := store.Job(context.Background(), claim.Job.ID) + if err != nil || job.State != protocol.JobFailed { + t.Fatalf("job = %q (err %v), want failed", job.State, err) + } + if _, err := store.RetryPublish(context.Background(), claim.Job.ID, + protocol.PublishRetryRequest{WorkerID: workerA, LeaseToken: tokenB}); err == nil || + serviceCode(t, err) != "publish_retry_not_allowed" { + t.Fatalf("publish retry of a lost attempt: err=%v, want publish_retry_not_allowed", err) + } +} + +// Migration 006 is additive: rows written under 005 keep every column, and +// read back unmarked. +func TestThePublishRetryMarkerMigrationKeepsExistingRows(t *testing.T) { + path := t.TempDir() + "/jig.db" + db, err := sql.Open("sqlite", "file:"+path+"?_pragma=foreign_keys(1)") + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(`CREATE TABLE schema_migrations (version INTEGER PRIMARY KEY, applied_at INTEGER NOT NULL)`); err != nil { + t.Fatal(err) + } + entries, err := migrations.Files.ReadDir(".") + if err != nil { + t.Fatal(err) + } + sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() }) + for _, entry := range entries { + version, err := migrationVersion(entry.Name()) + if err != nil || version >= 6 { + continue + } + body, err := migrations.Files.ReadFile(entry.Name()) + if err != nil { + t.Fatal(err) + } + if _, err := db.Exec(string(body)); err != nil { + t.Fatalf("apply %s: %v", entry.Name(), err) + } + if _, err := db.Exec(`INSERT INTO schema_migrations(version, applied_at) VALUES (?, 0)`, version); err != nil { + t.Fatal(err) + } + } + jobID := seedJob(t, db) + if _, err := db.Exec(` + INSERT INTO attempts(id, job_id, worker_id, attempt_number, state, result, error, created_at) + VALUES ('attempt-1', ?, 'worker-1', 1, 'accepted_unpublished', '{"kept":true}', 'gh_failed', 7) + `, jobID); err != nil { + t.Fatal(err) + } + + if err := Migrate(context.Background(), db); err != nil { + t.Fatalf("migrate to 006: %v", err) + } + var state, result, errorText string + var createdAt int64 + var marker int + if err := db.QueryRow(`SELECT state, result, error, created_at, publish_retry FROM attempts WHERE id = 'attempt-1'`). + Scan(&state, &result, &errorText, &createdAt, &marker); err != nil { + t.Fatalf("read migrated row: %v", err) + } + if state != "accepted_unpublished" || result != `{"kept":true}` || errorText != "gh_failed" || + createdAt != 7 || marker != 0 { + t.Fatalf("migrated row = %s %s %s %d marker=%d, want it unchanged and unmarked", + state, result, errorText, createdAt, marker) + } + if _, err := db.Exec(`UPDATE attempts SET publish_retry = 2 WHERE id = 'attempt-1'`); err == nil { + t.Fatal("publish_retry accepted a value other than 0 or 1") + } +} diff --git a/internal/controlplane/store.go b/internal/controlplane/store.go index 8f8822e..a3e6ac6 100644 --- a/internal/controlplane/store.go +++ b/internal/controlplane/store.go @@ -574,7 +574,7 @@ func (s *Store) CompleteAttempt(ctx context.Context, attemptID string, input pro } } if _, err := tx.ExecContext(ctx, ` - UPDATE attempts SET state = ?, result = ?, error = ?, completed_at = ? + UPDATE attempts SET state = ?, result = ?, error = ?, completed_at = ?, publish_retry = 0 WHERE id = ? AND state IN ('preparing', 'running') `, input.State, nullString(input.Result), nullString(input.Error), now, attemptID); err != nil { return protocol.Attempt{}, unavailable(err) diff --git a/internal/controlplane/sweep.go b/internal/controlplane/sweep.go index f9b9d62..ccd0eb3 100644 --- a/internal/controlplane/sweep.go +++ b/internal/controlplane/sweep.go @@ -20,10 +20,13 @@ import ( "github.com/StructuPath/jig/internal/protocol" ) -// LostAttempt names one attempt the sweeper marked lost and its job. +// LostAttempt names one attempt the sweeper marked lost and its job. State is +// the state the sweep left it in: `lost`, or `accepted_unpublished` for an +// interrupted publish-only retry, which keeps its attempt. type LostAttempt struct { AttemptID string JobID string + State string } // Sweeper owns the missed-heartbeat bookkeeping. It is not persistent by @@ -62,7 +65,7 @@ func (sw *Sweeper) Run(ctx context.Context, logger *slog.Logger) { for _, value := range lost { logger.Info("state_change", "resource_type", "attempt", "resource_id", value.AttemptID, - "job_id", value.JobID, "new_state", "lost") + "job_id", value.JobID, "new_state", value.State) } } } @@ -146,21 +149,32 @@ func (sw *Sweeper) Tick(ctx context.Context) ([]LostAttempt, error) { // The guarded UPDATE re-checks state and expiry inside the transaction, so a // heartbeat that renewed the lease between our read and this write wins and // the sweep is a no-op (KTD3). +// +// An expired publish-only retry is the exception: it ran no phase, so there +// is nothing lost but the publish, and the attempt goes back to +// accepted_unpublished — same attempt, branch, and publish records — where +// the retry is available again (R14). Its lease digest is cleared, so the +// interrupted publisher, should it wake, can neither heartbeat, record, nor +// complete. func (s *Store) sweepAttempt(ctx context.Context, attemptID string, nowMillis int64) (LostAttempt, bool, error) { - value := LostAttempt{AttemptID: attemptID} + value := LostAttempt{AttemptID: attemptID, State: protocol.AttemptLost} tx, err := s.db.BeginTx(ctx, nil) if err != nil { return value, false, unavailable(err) } defer tx.Rollback() + var publishRetry bool if err := tx.QueryRowContext(ctx, ` - SELECT job_id FROM attempts WHERE id = ? - `, attemptID).Scan(&value.JobID); err != nil { + SELECT job_id, publish_retry FROM attempts WHERE id = ? + `, attemptID).Scan(&value.JobID, &publishRetry); err != nil { return value, false, unavailable(err) } + if publishRetry { + return s.returnInterruptedPublishRetry(ctx, tx, value, nowMillis) + } result, err := tx.ExecContext(ctx, ` UPDATE attempts SET state = 'lost', error = 'lease expired: consecutive heartbeats missed', completed_at = ? - WHERE id = ? AND state IN ('preparing', 'running') AND lease_expires_at <= ? + WHERE id = ? AND state IN ('preparing', 'running') AND publish_retry = 0 AND lease_expires_at <= ? `, nowMillis, attemptID, nowMillis) if err != nil { return value, false, unavailable(err) @@ -188,6 +202,42 @@ func (s *Store) sweepAttempt(ctx context.Context, attemptID string, nowMillis in return value, true, nil } +// returnInterruptedPublishRetry is sweepAttempt's publish-retry half, inside +// the same transaction and under the same guard: a renewal or completion +// that landed first wins and the sweep is a no-op. +func (s *Store) returnInterruptedPublishRetry( + ctx context.Context, tx *sql.Tx, value LostAttempt, nowMillis int64, +) (LostAttempt, bool, error) { + value.State = protocol.AttemptAcceptedUnpublished + result, err := tx.ExecContext(ctx, ` + UPDATE attempts + SET state = 'accepted_unpublished', publish_retry = 0, lease_digest = NULL, completed_at = ?, + error = 'publish retry interrupted: lease expired; the publish-only retry is available again' + WHERE id = ? AND state = 'running' AND publish_retry = 1 AND lease_expires_at <= ? + `, nowMillis, value.AttemptID, nowMillis) + if err != nil { + return value, false, unavailable(err) + } + if changed, _ := result.RowsAffected(); changed != 1 { + if err := tx.Commit(); err != nil { + return value, false, unavailable(err) + } + return value, false, nil + } + if _, err := tx.ExecContext(ctx, ` + UPDATE jobs SET state = 'accepted_unpublished', updated_at = ? WHERE id = ? AND state = 'active' + `, nowMillis, value.JobID); err != nil { + return value, false, unavailable(err) + } + if err := applyRunAggregationForJob(ctx, tx, value.JobID, nowMillis); err != nil { + return value, false, err + } + if err := tx.Commit(); err != nil { + return value, false, unavailable(err) + } + return value, true, nil +} + // Heartbeat renews an attempt's lease in one fenced transaction (R5, R6) and // carries cancellation back to the worker. The token digest must own the // attempt, and renewal is permitted — expired or not — iff the attempt has diff --git a/internal/engine/code_phase.go b/internal/engine/code_phase.go index efbb69b..aa7343b 100644 --- a/internal/engine/code_phase.go +++ b/internal/engine/code_phase.go @@ -34,6 +34,11 @@ type commandResult struct { // reports_fields report longer than the tail still parses. LastLine string LastLineOverflow bool + // RecordError is set when the command's process group could not be + // recorded as live. The command was killed and reaped before it could + // do anything worth judging, so this is an infrastructure failure, never + // a verdict on the command. + RecordError string } // Passed reports a clean exit. @@ -45,6 +50,17 @@ func (r commandResult) Passed() bool { // the given environment and timeout. The command gets its own process group // so a timeout kills everything it spawned, not just the shell. // +// record, when non-nil, is told the group is live right after the start and +// no longer live once the shell has exited or been killed and reaped — the +// same record agent sends keep (U4), so a worker that dies hard mid-command +// leaves a manifest its restart reconciliation can stop the group from. A +// command whose group cannot be recorded does not run on unrecorded: the +// group is killed, reaped, and the record cleared again in case the failed +// write half-landed. Only the group is tracked, through its leader: a +// background process the command leaves running after a normal exit is +// left alone exactly as before, and once its leader is gone reconciliation +// cannot verify the group as ours, so it is not signalled. +// // env is REQUIRED. A nil env means Go hands the child jig's own os.Environ() // — every API key the operator exported, and their real HOME — which is // exactly the containment KTD10 and KTD11 exist to remove. Commands run in @@ -52,6 +68,7 @@ func (r commandResult) Passed() bool { // code), so inheritance must fail loudly rather than silently return. func runShellCommand( ctx context.Context, dir, command string, env []string, timeout time.Duration, + record func(groupID int64, active bool) error, ) commandResult { if env == nil { return commandResult{ExitCode: -1, StartError: "refusing to run a command with an " + @@ -74,6 +91,16 @@ func runShellCommand( } done := make(chan error, 1) go func() { done <- shell.Wait() }() + groupID := int64(shell.Process.Pid) + if record != nil { + if err := record(groupID, true); err != nil { + _ = syscall.Kill(-shell.Process.Pid, syscall.SIGKILL) + <-done + _ = record(groupID, false) + return commandResult{ExitCode: -1, RecordError: err.Error()} + } + defer func() { _ = record(groupID, false) }() + } select { case err := <-done: last, overflow := lines.Last() diff --git a/internal/engine/command_record_test.go b/internal/engine/command_record_test.go new file mode 100644 index 0000000..f6fc0d6 --- /dev/null +++ b/internal/engine/command_record_test.go @@ -0,0 +1,320 @@ +// command_record_test.go — jig-side commands record their process groups +// like agent sends do (U4): a code phase's command and a tests_pass gate's +// are recorded live while they run and cleared once they have exited, so a +// worker that dies hard mid-command leaves a manifest its restart can stop +// the group from. A group that cannot be recorded does not run on: it is +// killed, and the attempt fails as infrastructure — never as a failing test +// that a repair edge or a gate correction would spend model budget on. +package engine_test + +import ( + "context" + "errors" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/StructuPath/jig/internal/engine" + "github.com/StructuPath/jig/internal/engine/enginetest" + "github.com/StructuPath/jig/internal/protocol" + "github.com/StructuPath/jig/internal/worker" +) + +// groupRecorder is a RecordProcess fake: the live set per attempt, and an +// optional failure for every active record. +type groupRecorder struct { + mutex sync.Mutex + live map[int64]bool + calls []string + failure error +} + +func (r *groupRecorder) record(attemptID string, groupID int64, active bool) error { + r.mutex.Lock() + defer r.mutex.Unlock() + if attemptID != "attempt-1" { + return errors.New("recorded under the wrong attempt: " + attemptID) + } + r.calls = append(r.calls, strconv.FormatInt(groupID, 10)+"="+strconv.FormatBool(active)) + if active { + // A failed write may still have half-landed: the group is recorded + // either way, so only a clear afterwards leaves the set honest. + r.live[groupID] = true + return r.failure + } + delete(r.live, groupID) + return nil +} + +func (r *groupRecorder) isLive(groupID int64) bool { + r.mutex.Lock() + defer r.mutex.Unlock() + return r.live[groupID] +} + +func (r *groupRecorder) liveCount() int { + r.mutex.Lock() + defer r.mutex.Unlock() + return len(r.live) +} + +// heldCommand is a shell command that writes its own process id — its +// group's id, since it leads its group — and then waits for release. +func heldCommand(t *testing.T) (command, pidFile, release string) { + t.Helper() + dir := t.TempDir() + pidFile = filepath.Join(dir, "pid") + release = filepath.Join(dir, "release") + command = "echo $$ > '" + pidFile + "'; while [ ! -f '" + release + "' ]; do sleep 0.05; done" + // A failing assertion must not leave the loop running: release it, and + // stop THIS command's own group — never 0, 1, or the test's own. + t.Cleanup(func() { + _ = os.WriteFile(release, nil, 0o644) + body, err := os.ReadFile(pidFile) + if err != nil { + return + } + if group, err := strconv.Atoi(strings.TrimSpace(string(body))); err == nil && + group > 1 && group != syscall.Getpgrp() { + _ = syscall.Kill(-group, syscall.SIGKILL) + } + }) + return command, pidFile, release +} + +func awaitPID(t *testing.T, pidFile string) int64 { + t.Helper() + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + if body, err := os.ReadFile(pidFile); err == nil && strings.HasSuffix(string(body), "\n") { + pid, err := strconv.ParseInt(strings.TrimSpace(string(body)), 10, 64) + if err != nil { + t.Fatal(err) + } + return pid + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("the command never started") + return 0 +} + +// Both jig-side command sites record their group live while it runs and +// clear it once it has exited. +func TestCodeAndGateCommandsAreRecordedLiveWhileTheyRun(t *testing.T) { + for name, phases := range map[string]func(command string) string{ + "code phase": func(command string) string { + return " - name: verify\n kind: code\n command: \"" + command + "\"\n" + }, + "tests_pass gate": func(command string) string { + return " - name: build\n kind: agent\n owner: writer\n gates:\n" + + " - {name: tests_pass, command: \"" + command + "\"}\n" + }, + } { + t.Run(name, func(t *testing.T) { + repo := initRepo(t) + recorder := &groupRecorder{live: map[int64]bool{}} + fake := enginetest.New(enginetest.Step{Text: envelope(map[string]any{"status": "success", "summary": "done"})}) + runner := newTestRunner(t, fake, &recordingSink{}, func(config *engine.Config) { + config.RecordProcess = recorder.record + }) + command, pidFile, release := heldCommand(t) + snapshot := "name: t\n" + oneWriterRoster + "phases:\n" + phases(command) + "acceptance: [all_phases_passed]\n" + + outcomes := make(chan worker.Outcome, 1) + go func() { outcomes <- runner.Execute(context.Background(), testAttempt(snapshot, nil, repo)) }() + pid := awaitPID(t, pidFile) + if !recorder.isLive(pid) { + t.Fatalf("the running command's group %d is not recorded live (calls %v)", pid, recorder.calls) + } + if err := os.WriteFile(release, nil, 0o644); err != nil { + t.Fatal(err) + } + outcome := <-outcomes + if outcome.State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("state = %q (%s), want accepted_unpublished", outcome.State, outcome.Error) + } + if recorder.liveCount() != 0 { + t.Fatalf("groups still recorded live after the command exited: calls %v", recorder.calls) + } + }) + } +} + +// A group that cannot be recorded is killed before it does anything worth +// judging — the delayed marker never appears — and is cleared again. The +// attempt fails as infrastructure: the code phase's repair edge is not +// followed, and the gate sends the agent no correction. +func TestACommandWhoseGroupCannotBeRecordedIsKilledAndFailsAsInfrastructure(t *testing.T) { + for name, phases := range map[string]func(command string) string{ + "code phase": func(command string) string { + return " - {name: build, kind: agent, owner: writer}\n" + + " - name: verify\n kind: code\n command: \"" + command + "\"\n" + + " on_fail: {run: build, then: rerun-self, budget: 2}\n" + }, + "tests_pass gate": func(command string) string { + return " - name: build\n kind: agent\n owner: writer\n gates:\n" + + " - {name: tests_pass, command: \"" + command + "\"}\n" + }, + } { + t.Run(name, func(t *testing.T) { + repo := initRepo(t) + recorder := &groupRecorder{live: map[int64]bool{}, failure: errors.New("manifest write failed")} + done := envelope(map[string]any{"status": "success", "summary": "done"}) + fake := enginetest.New(enginetest.Step{Text: done}, enginetest.Step{Text: done}, enginetest.Step{Text: done}) + sink := &recordingSink{} + runner := newTestRunner(t, fake, sink, func(config *engine.Config) { + config.RecordProcess = recorder.record + }) + marker := filepath.Join(t.TempDir(), "ran") + command := "sleep 0.5; touch '" + marker + "'" + snapshot := "name: t\n" + oneWriterRoster + "phases:\n" + phases(command) + "acceptance: [all_phases_passed]\n" + + outcome := runner.Execute(context.Background(), testAttempt(snapshot, nil, repo)) + if outcome.State != protocol.AttemptFailed || !strings.Contains(outcome.Error, "manifest write failed") { + t.Fatalf("state = %q (%s), want failed naming the record failure", outcome.State, outcome.Error) + } + if calls := len(fake.Calls()); calls != 1 { + t.Fatalf("the runtime was called %d times, want 1: no repair and no gate correction", calls) + } + if sink.has(protocol.EventLog, "repair_edge") { + t.Fatal("a record failure was routed through the repair edge") + } + if !sink.has(protocol.EventError, "engine_error") && name == "code phase" { + t.Fatal("the code phase's record failure was not traced as an engine error") + } + time.Sleep(time.Second) + if _, err := os.Stat(marker); !os.IsNotExist(err) { + t.Fatalf("the unrecorded command kept running and wrote its marker (stat: %v)", err) + } + if recorder.liveCount() != 0 { + t.Fatalf("the failed record was never cleared: calls %v", recorder.calls) + } + }) + } +} + +// An agent phase whose own verdict fires its on_fail edge must still not be +// repaired when its tests_pass gate could not run: the record failure is +// jig's, not the agent's, so it spends neither a gate correction nor the +// repair budget — serially or inside a parallel group. Each case first runs +// with recording working, which proves the edge is live: the same envelope +// does dispatch the repair there. +func TestAGateRecordFailureNeverFollowsTheAgentPhasesRepairEdge(t *testing.T) { + rejecting := enginetest.Step{Text: envelope(map[string]any{ + "status": "success", "summary": "not yet", "approved": false, "blocking": []any{"not yet"}, + })} + done := enginetest.Step{Text: envelope(map[string]any{"status": "success", "summary": "done"})} + serial := "name: t\n" + oneWriterRoster + `phases: + - {name: plan, kind: agent, owner: writer} + - name: build + kind: agent + owner: writer + gates: [{name: tests_pass, command: "true"}] + on_fail: {when: "approved == false", run: plan, then: rerun-self, budget: 2, exhausted: fail-job} +acceptance: [all_phases_passed] +` + grouped := strings.Replace(panelSnapshot(nil, panelGroup), + " owner: security\n gates: [{name: verdict_consistent}]\n", + " owner: security\n gates: [{name: verdict_consistent}, {name: tests_pass, command: \"true\"}]\n", 1) + if grouped == panelSnapshot(nil, panelGroup) { + t.Fatal("the panel fixture changed; the tests_pass gate was not added to a member") + } + + run := func(t *testing.T, snapshot string, failure error, route func(*enginetest.Runtime)) (*repairFixture, worker.Outcome) { + recorder := &groupRecorder{live: map[int64]bool{}, failure: failure} + f := newRepairFixture(t, snapshot, func(config *engine.Config) { + config.RecordProcess = recorder.record + }, done, rejecting, done, rejecting, done, rejecting) + if route != nil { + route(f.fake) + } + return f, executeWithin(t, f, 30*time.Second) + } + routePanel := func(fake *enginetest.Runtime) { + fake.Route(buildRole, writes(map[string]string{"src/app.txt": "v1"}, "built"), + writes(map[string]string{"src/app.txt": "v2"}, "rebuilt")) + fake.Route(correctnessRole, approve("correct", nil), approve("correct", nil)) + fake.Route(securityRole, reject("auth bypass"), approve("secure now", nil)) + fake.Route(maintainabilityRole, approve("maintainable", nil), approve("maintainable", nil)) + fake.Route(closerRole, success("wrapped up", nil)) + } + for name, tc := range map[string]struct { + snapshot string + route func(*enginetest.Runtime) + phase string + calls int // runtime calls up to and including the failing phase + }{ + "serial": {serial, nil, "build", 2}, + "parallel member": {grouped, routePanel, "review-security", 4}, + } { + t.Run(name, func(t *testing.T) { + control, _ := run(t, tc.snapshot, nil, tc.route) + if !control.sink.has(protocol.EventLog, "repair_edge") { + t.Fatal("control: with recording working the edge never fired; the fixture proves nothing") + } + + f, outcome := run(t, tc.snapshot, errors.New("manifest write failed"), tc.route) + if outcome.State != protocol.AttemptFailed || !strings.Contains(outcome.Error, "manifest write failed") { + t.Fatalf("state = %q (%s), want failed naming the record failure", outcome.State, outcome.Error) + } + if f.sink.has(protocol.EventLog, "repair_edge") { + t.Fatal("a gate record failure dispatched the agent phase's repair edge") + } + if calls := len(f.fake.Calls()); calls != tc.calls { + t.Fatalf("runtime calls = %d, want %d: no correction send and no repair dispatch", calls, tc.calls) + } + ends, failedEnds := 0, 0 + for _, event := range f.sink.all() { + if event.Phase != tc.phase { + continue + } + if event.Type == protocol.EventAgentEnd { + ends++ + } + if event.Type == protocol.EventPhaseEnd && strings.Contains(string(event.Payload), protocol.EnvelopeFail) { + failedEnds++ + } + } + if ends != 1 || failedEnds != 1 { + t.Fatalf("phase %s closed with %d agent_end and %d failed phase_end, want exactly one each", + tc.phase, ends, failedEnds) + } + }) + } +} + +// That exit is still a live agent's exit: the write boundary is enforced on +// it, so an out-of-allowlist write is rolled back and ends the attempt as a +// breach rather than surviving into the retained worktree. +func TestAGateRecordFailureStillEnforcesTheWriteBoundary(t *testing.T) { + snapshot := "name: t\n" + oneWriterRoster + `phases: + - name: build + kind: agent + owner: writer + gates: [{name: tests_pass, command: "true"}] +acceptance: [all_phases_passed] +` + recorder := &groupRecorder{live: map[int64]bool{}, failure: errors.New("manifest write failed")} + breaching := enginetest.Step{Files: map[string]string{"evil.txt": "planted"}, + Text: envelope(map[string]any{"status": "success", "summary": "done"})} + f := newRepairFixture(t, snapshot, func(config *engine.Config) { + config.RecordProcess = recorder.record + }, breaching, breaching) + outcome := executeWithin(t, f, 30*time.Second) + if outcome.State != protocol.AttemptFailed || !strings.Contains(outcome.Error, "manifest write failed") || + !f.sink.has(protocol.EventError, "write_boundary_breach") { + t.Fatalf("state = %q (%s), want a failure naming the record failure and the breach", outcome.State, outcome.Error) + } + if _, err := os.Stat(filepath.Join(f.repo, "evil.txt")); !os.IsNotExist(err) { + t.Fatalf("the out-of-allowlist write survived (stat: %v)", err) + } + if calls := len(f.fake.Calls()); calls != 1 { + t.Fatalf("runtime calls = %d, want 1", calls) + } +} diff --git a/internal/engine/gates.go b/internal/engine/gates.go index a561a5f..ebe72fc 100644 --- a/internal/engine/gates.go +++ b/internal/engine/gates.go @@ -38,31 +38,37 @@ type gateContext struct { // environment and the operator's real HOME. env []string timeout timeoutConfig + // record reports a gate command's process group live and gone, as the + // code phase does (runShellCommand). Nil records nothing. + record func(groupID int64, active bool) error } // runGate dispatches one configured gate by registry name. Names were // resolved at save time (R1); an unknown name here is a frozen-snapshot // integrity failure, reported as a failed check rather than a panic. -func runGate(gc gateContext, spec protocol.GateSpec) protocol.GateReport { +// +// The error is jig's own failure to run the gate, not the gate's verdict: it +// is no claim the agent made, so it must never become a correction prompt. +func runGate(gc gateContext, spec protocol.GateSpec) (protocol.GateReport, error) { gc.command = spec.Command gc.allow = spec.Allow switch spec.Name { case protocol.GateArtifactsExist: - return gateArtifactsExist(gc) + return gateArtifactsExist(gc), nil case protocol.GateFilesNonEmpty: - return gateFilesNonEmpty(gc) + return gateFilesNonEmpty(gc), nil case protocol.GateDiffMatchesClaims: - return gateDiffMatchesClaims(gc) + return gateDiffMatchesClaims(gc), nil case protocol.GateVerdictConsistent: - return gateVerdictConsistent(gc) + return gateVerdictConsistent(gc), nil case protocol.GateTestsPass: return gateTestsPass(gc) case protocol.GateTestsIntact: - return gateTestsIntact(gc) + return gateTestsIntact(gc), nil } var report protocol.GateReport report.Check(spec.Name, false, "gate is not in the built-in registry") - return report + return report, nil } // artifactPath resolves a claimed artifact against the worktree, rejecting @@ -236,9 +242,13 @@ func gateVerdictConsistent(gc gateContext) protocol.GateReport { // gateTestsPass: the configured command must exit 0 in the worktree. The // note carries the exit status and, on failure, a bounded output tail as // evidence the agent can act on. -func gateTestsPass(gc gateContext) protocol.GateReport { +func gateTestsPass(gc gateContext) (protocol.GateReport, error) { var report protocol.GateReport - result := runShellCommand(gc.ctx, gc.worktree, gc.command, gc.env, gc.timeout.phase) + result := runShellCommand(gc.ctx, gc.worktree, gc.command, gc.env, gc.timeout.phase, gc.record) + if result.RecordError != "" { + return report, fmt.Errorf("gate %s: record the command's process group: %s", + protocol.GateTestsPass, result.RecordError) + } note := fmt.Sprintf("exit %d", result.ExitCode) if result.TimedOut { note = "timed out" @@ -247,5 +257,5 @@ func gateTestsPass(gc gateContext) protocol.GateReport { note += "\n" + result.OutputTail } report.Check(gc.command, result.Passed(), note) - return report + return report, nil } diff --git a/internal/engine/phase.go b/internal/engine/phase.go index da65cbe..b75d426 100644 --- a/internal/engine/phase.go +++ b/internal/engine/phase.go @@ -913,18 +913,22 @@ func (e *execution) phaseCorrectionBudget(phase protocol.PhaseSpec) int { // ALL gates, never only the failed one (R9). func (e *execution) runGates( ctx context.Context, phase protocol.PhaseSpec, envelope parsedEnvelope, attempt int, -) protocol.GateReport { +) (protocol.GateReport, error) { var merged protocol.GateReport env := e.phaseEnv(phase) for _, gate := range phase.Gates { - report := runGate(gateContext{ + report, err := runGate(gateContext{ ctx: ctx, worktree: e.attempt.WorktreePath, envelope: envelope, baseSHA: e.attempt.BaseSHA, env: env, timeout: e.timeouts, + record: e.commandRecorder(), }, gate) + if err != nil { + return merged, err + } e.gateReports[gate.Name] = report merged.Checks = append(merged.Checks, report.Checks...) eventType := protocol.EventGatePass @@ -937,7 +941,27 @@ func (e *execution) runGates( "violations": report.Violations(), }) } - return merged + return merged, nil +} + +// commandRecorder is the record callback for jig-side commands — a code +// phase's command and a tests_pass gate's — into the same manifest set agent +// process groups go to (U4). Nil when nothing records. A failed clear is +// logged exactly as an agent's is; a failed start is returned, and the +// command is killed rather than run unrecorded (runShellCommand). +func (e *execution) commandRecorder() func(groupID int64, active bool) error { + if e.runner.config.RecordProcess == nil { + return nil + } + attemptID := e.attempt.Claim.Attempt.ID + return func(groupID int64, active bool) error { + err := e.runner.config.RecordProcess(attemptID, groupID, active) + if err != nil { + e.runner.config.Logger.Warn("process_group_record_failed", + "attempt_id", attemptID, "process_group_id", groupID, "active", active, "error", err) + } + return err + } } // phaseEnv composes the environment for everything this phase runs jig-side: @@ -1011,11 +1035,17 @@ func (e *execution) runCodePhase(ctx context.Context, phase protocol.PhaseSpec) case <-commandCtx.Done(): } }() - result := runShellCommand(commandCtx, e.attempt.WorktreePath, phase.Command, env, e.timeouts.phase) + result := runShellCommand(commandCtx, e.attempt.WorktreePath, phase.Command, env, e.timeouts.phase, + e.commandRecorder()) stopCommand() if e.cancelled() { return phaseRun{outcome: phaseCancelled, failure: "cancelled during phase " + phase.Name} } + if result.RecordError != "" { + // Not the command's failure: it was stopped before it could earn a + // verdict, so no repair edge may treat it as a failing test. + return e.failInfra(phase.Name, "record the command's process group: "+result.RecordError) + } var reported map[string]any if phase.ReportsFields && result.Passed() { fields, err := reportedFields(result.LastLine, result.LastLineOverflow) @@ -1036,7 +1066,10 @@ func (e *execution) runCodePhase(ctx context.Context, phase protocol.PhaseSpec) // Gates verify the adapter envelope's claims too; there is no session to // correct, so a violation is a phase failure that the repair edge (if // declared) routes like any other code-phase failure. - report := e.runGates(ctx, phase, envelope, entry) + report, gateErr := e.runGates(ctx, phase, envelope, entry) + if gateErr != nil { + return e.failInfra(phase.Name, gateErr.Error()) + } passed := result.Passed() && report.Passed() status := protocol.EnvelopeFail if passed { @@ -1318,7 +1351,12 @@ func (e *execution) runAgentPhaseAttempt( return fail(parseDetail) } - report := e.runGates(ctx, phase, envelope, gateAttempt) + report, gateErr := e.runGates(ctx, phase, envelope, gateAttempt) + if gateErr != nil { + // jig could not run the gate: there is nothing for the agent to + // correct, so this ends the phase without a correction send. + return fail(gateErr.Error()) + } if report.Passed() { break } diff --git a/internal/worker/publish_retry_recovery_test.go b/internal/worker/publish_retry_recovery_test.go new file mode 100644 index 0000000..ac9f0f8 --- /dev/null +++ b/internal/worker/publish_retry_recovery_test.go @@ -0,0 +1,125 @@ +// publish_retry_recovery_test.go — a publish-only retry that stalls (CI that +// never answers, a process that stops heartbeating) loses its lease to the +// sweeper, which returns the attempt to accepted_unpublished rather than +// `lost`. The stalled retry can no longer write anything when it wakes, and +// the next retry re-enters the SAME attempt: same branch and pull request, CI +// judged on the branch's current head, and still not one phase re-run. +package worker + +import ( + "context" + "slices" + "sync" + "testing" + "time" + + "github.com/StructuPath/jig/internal/controlplane" + "github.com/StructuPath/jig/internal/protocol" +) + +func TestAStalledPublishRetryIsSweptBackAndTheNextRetryFinishesTheSameAttempt(t *testing.T) { + h := newHarness(t) + originDir, head, identity := newOriginRepo(t) + h.seedRunWithSnapshot("run-1", ciSnapshot, protocol.RunTarget{Repository: identity, BaseSHA: head}) + job := h.enqueue("run-1", identity) + branch := protocol.PublishBranch(job.ID, 1) + + var phaseRuns int + inner := RunnerFunc(func(ctx context.Context, attempt *PreparedAttempt) Outcome { + phaseRuns++ + return writeAndDeclare(t, map[string]string{"work.txt": "work\n"}).Run(ctx, attempt) + }) + // CI is red for the attempt, never answers for the first retry, and is + // green for the second. + var mode string + entered, release := make(chan struct{}), make(chan struct{}) + var enterOnce sync.Once + gateway := newFakeGateway() + gateway.checks = func(string, int) ([]CICheck, error) { + switch mode { + case "stall": + enterOnce.Do(func() { close(entered) }) + <-release + return []CICheck{check("test", CIPass)}, nil + case "green": + return []CICheck{check("test", CIPass)}, nil + } + return []CICheck{check("test", CIFail)}, nil + } + w := newCIWorker(t, h, inner, gateway) + attempt, err := w.ClaimOnce(context.Background()) + if err != nil || attempt == nil || attempt.State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("claim = %+v (%v), want accepted_unpublished on red CI", attempt, err) + } + + mode = "stall" + stalled := make(chan error, 1) + go func() { + _, err := w.RetryPublish(context.Background(), job.ID, gateway, fastCIOptions()) + stalled <- err + }() + select { + case <-entered: + case <-time.After(30 * time.Second): + t.Fatal("the first retry never reached CI") + } + // Its lease runs out while CI hangs: three missed heartbeats at the + // sweeper's cadence, compressed by expiring the lease outright. + if _, err := h.db.Exec(`UPDATE attempts SET lease_expires_at = ? WHERE id = ?`, + time.Now().Add(-time.Minute).UnixMilli(), attempt.ID); err != nil { + t.Fatal(err) + } + sweeper := controlplane.NewSweeper(h.store) + var swept []controlplane.LostAttempt + for i := 0; i < protocol.MissedHeartbeatsBeforeSweep; i++ { + ticked, err := sweeper.Tick(context.Background()) + if err != nil { + t.Fatal(err) + } + swept = append(swept, ticked...) + } + if len(swept) != 1 || swept[0].State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("sweep = %+v, want the retry's attempt returned to accepted_unpublished", swept) + } + + // A person pushes a fix while the stalled retry is still asleep; then it + // wakes, judges green, and is fenced out of every write. + fixed := pushToBranch(t, originDir, branch) + close(release) + select { + case err := <-stalled: + if err == nil { + t.Fatal("the swept retry completed its attempt") + } + case <-time.After(60 * time.Second): + t.Fatal("the stalled retry never returned") + } + if steps := recordedSteps(t, w, attempt.ID); slices.Contains(steps, protocol.PublishStepCI) { + t.Fatalf("ledger steps = %v: the swept retry recorded ci", steps) + } + if after, err := h.store.Attempt(context.Background(), attempt.ID); err != nil || + after.State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("attempt after the zombie woke = %+v (%v), want accepted_unpublished", after.State, err) + } + + mode = "green" + retried, err := w.RetryPublish(context.Background(), job.ID, gateway, fastCIOptions()) + if err != nil { + t.Fatalf("second retry: %v", err) + } + summary := publishSummaryOf(t, retried.Result) + if retried.ID != attempt.ID || retried.State != protocol.AttemptAccepted || summary.CIRef != fixed { + t.Fatalf("second retry = %s %s judged %s, want attempt %s accepted on the current head %s", + retried.ID, retried.State, summary.CIRef, attempt.ID, fixed) + } + if created, _ := gateway.counts(); created != 1 { + t.Fatalf("pull requests created = %d, want 1", created) + } + if phaseRuns != 1 || len(gateway.reruns) != 0 || len(summary.CIRepairs) != 0 { + t.Fatalf("phase runs %d, CI re-runs %v, repairs %v: a publish retry re-runs and repairs nothing", + phaseRuns, gateway.reruns, summary.CIRepairs) + } + if remoteBranches(t, originDir)[branch] != fixed { + t.Fatal("the branch moved off the person's fix") + } +} diff --git a/migrations/006_publish_retry_marker.sql b/migrations/006_publish_retry_marker.sql new file mode 100644 index 0000000..cd929e1 --- /dev/null +++ b/migrations/006_publish_retry_marker.sql @@ -0,0 +1,12 @@ +-- 006_publish_retry_marker.sql — the publish-only retry is marked on its +-- attempt. +-- +-- A publish-only retry re-leases an accepted_unpublished attempt into +-- `running` (R14). Without a marker, an interrupted retry's expired lease is +-- indistinguishable from an interrupted phase run, so the sweeper marks it +-- `lost` and the job's only road left is a cold retry: a new attempt, a new +-- branch, a second pull request. The flag says "this lease is a publish +-- retry": the sweeper returns such an attempt to accepted_unpublished, same +-- attempt and records, instead. Additive, defaulting every existing row to 0. + +ALTER TABLE attempts ADD COLUMN publish_retry INTEGER NOT NULL DEFAULT 0 CHECK (publish_retry IN (0, 1)); From b05ee7717600a9490bb5fc7cd2df77e0a8d0ada1 Mon Sep 17 00:00:00 2001 From: Victor Garcia Date: Thu, 29 Oct 2026 22:14:01 -0600 Subject: [PATCH 2/2] fix(worker): harden recovery locking and command cleanup --- cmd/jig/crash_reconcile_test.go | 15 ++- cmd/jig/publish_retry.go | 34 +++++- cmd/jig/publish_retry_test.go | 44 ++++++++ docs/dogfooding.md | 2 +- docs/quickstart.md | 2 +- internal/engine/code_phase.go | 9 +- internal/engine/command_record_test.go | 147 ++++++++++++++++++++----- internal/engine/phase.go | 13 ++- 8 files changed, 224 insertions(+), 42 deletions(-) diff --git a/cmd/jig/crash_reconcile_test.go b/cmd/jig/crash_reconcile_test.go index ddc2584..48eab26 100644 --- a/cmd/jig/crash_reconcile_test.go +++ b/cmd/jig/crash_reconcile_test.go @@ -31,8 +31,10 @@ func TestARestartedWorkerStopsTheCodePhaseGroupAHardKilledWorkerLeftRunning(t *t defer stopServe() pidFile, heartbeat := filepath.Join(scratch, "pid"), filepath.Join(scratch, "heartbeat") + release := filepath.Join(scratch, "release") source := "name: crash-fixture\nphases:\n - name: loop\n kind: code\n command: |\n" + - " echo $$ > '" + pidFile + "'; while :; do echo beat >> '" + heartbeat + "'; sleep 0.05; done\n" + + " echo $$ > '" + pidFile + "'; while [ -d '" + scratch + "' ] && [ ! -e '" + release + "' ]; do " + + "echo beat >> '" + heartbeat + "'; sleep 0.05; done\n" + "acceptance: [all_phases_passed]\n" var definition protocol.Definition if err := apiCall(context.Background(), base, "POST", "/api/definitions", @@ -45,16 +47,13 @@ func TestARestartedWorkerStopsTheCodePhaseGroupAHardKilledWorkerLeftRunning(t *t t.Fatal(err) } + // Defensive cleanup releases the loop through its own file rather than + // signalling a cached group id, which the OS may have reused by then. + // Registered before the worker starts so an early failure releases it too. + t.Cleanup(func() { _ = os.WriteFile(release, nil, 0o600) }) child, _ := startHelper(t, "jig", "JIG_TEST_HELPER_ARGS="+strings.Join( []string{"worker", "--server", base, "--data", workerData}, "\x1f")) group := awaitCommandGroup(t, pidFile) - // Defensive cleanup of THIS test's group only: the id came from the - // shell this test's definition started, and it is never 0, 1, or ours. - t.Cleanup(func() { - if group > 1 && group != int64(syscall.Getpgrp()) { - _ = syscall.Kill(-int(group), syscall.SIGKILL) - } - }) awaitManifestGroup(t, workerData, group) if err := child.Process.Signal(syscall.SIGKILL); err != nil { diff --git a/cmd/jig/publish_retry.go b/cmd/jig/publish_retry.go index 83921fa..93776fb 100644 --- a/cmd/jig/publish_retry.go +++ b/cmd/jig/publish_retry.go @@ -193,10 +193,25 @@ type workerDataLock struct{ file *os.File } func lockWorkerData(dataDir string) (*workerDataLock, error) { path := filepath.Join(dataDir, "worker.lock") - file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) + // The lock file is written (the pid note below), so it must be the data + // directory's own: a worker.lock that is a symlink or a hard link to some + // other file would have that file truncated. O_NOFOLLOW refuses a symlink + // at open — a dangling one included, which O_CREAT would otherwise create + // the target of — and the link count, read from the open file, refuses a + // hard link. Both are refused, never repaired: the operator decides. + fd, err := syscall.Open(path, syscall.O_CREAT|syscall.O_RDWR|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0o600) + if errors.Is(err, syscall.ELOOP) { + return nil, fmt.Errorf("worker data lock %s is a symbolic link; jig refuses to follow it — "+ + "remove it and jig will create the lock file again", path) + } if err != nil { return nil, fmt.Errorf("open worker data lock: %w", err) } + file := os.NewFile(uintptr(fd), path) + if err := requireOwnLockFile(file, path); err != nil { + file.Close() + return nil, err + } if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { file.Close() if errors.Is(err, syscall.EWOULDBLOCK) { @@ -212,6 +227,23 @@ func lockWorkerData(dataDir string) (*workerDataLock, error) { return &workerDataLock{file: file}, nil } +// requireOwnLockFile checks the OPEN file, so what is checked is what gets +// locked and written: a regular file with exactly one name. +func requireOwnLockFile(file *os.File, path string) error { + info, err := file.Stat() + if err != nil { + return fmt.Errorf("inspect worker data lock: %w", err) + } + if !info.Mode().IsRegular() { + return fmt.Errorf("worker data lock %s is not a regular file; remove it and jig will create it again", path) + } + if stat, ok := info.Sys().(*syscall.Stat_t); !ok || stat.Nlink != 1 { + return fmt.Errorf("worker data lock %s has more than one name (a hard link); jig refuses to write "+ + "through it — remove it and jig will create the lock file again", path) + } + return nil +} + // release drops the lock. The file stays: removing it would let a racing // opener lock an inode nobody else can find. func (l *workerDataLock) release() { diff --git a/cmd/jig/publish_retry_test.go b/cmd/jig/publish_retry_test.go index bc336d8..2b0f92d 100644 --- a/cmd/jig/publish_retry_test.go +++ b/cmd/jig/publish_retry_test.go @@ -399,3 +399,47 @@ func mustGetwd(t *testing.T) string { } return dir } + +// A worker.lock that is a symbolic or hard link to some other file is +// refused before anything is written through it: the pid note would +// otherwise truncate the other file. A dangling symlink is refused too, where +// O_CREAT would create its target. The link and its target stay as they are. +func TestALinkedWorkerLockIsRefusedAndItsTargetLeftIntact(t *testing.T) { + for name, link := range map[string]func(target, lock string) error{ + "symlink": os.Symlink, + "hard link": os.Link, + "dangling symlink": func(target, lock string) error { return os.Symlink(target+".missing", lock) }, + } { + t.Run(name, func(t *testing.T) { + dataDir := t.TempDir() + target := filepath.Join(t.TempDir(), "precious.txt") + const precious = "operator data, not a lock\n" + if err := os.WriteFile(target, []byte(precious), 0o600); err != nil { + t.Fatal(err) + } + lockPath := filepath.Join(dataDir, "worker.lock") + if err := link(target, lockPath); err != nil { + t.Fatal(err) + } + + lock, err := lockWorkerData(dataDir) + if err == nil { + lock.release() + t.Fatal("a linked worker.lock was locked and written through") + } + if !strings.Contains(err.Error(), lockPath) || !strings.Contains(err.Error(), "remove it") { + t.Fatalf("refusal %q does not name the lock file and what to do", err) + } + code, _, stderr := publishRetry(t, "--server", "http://127.0.0.1:1", "--data", dataDir, "job-1") + if code != exitInfraFailed || !strings.Contains(stderr, "worker.lock") { + t.Fatalf("retry over a linked lock: exit %d, want %d naming worker.lock", code, exitInfraFailed) + } + if body, err := os.ReadFile(target); err != nil || string(body) != precious { + t.Fatalf("the link's target changed: %q (%v)", body, err) + } + if _, err := os.Stat(target + ".missing"); !os.IsNotExist(err) { + t.Fatalf("a dangling symlink's target was created (stat: %v)", err) + } + }) + } +} diff --git a/docs/dogfooding.md b/docs/dogfooding.md index d99978d..94c64d5 100644 --- a/docs/dogfooding.md +++ b/docs/dogfooding.md @@ -370,7 +370,7 @@ worktree and the repository cache are in that worker's data directory: ```sh # stop the worker that ran the job (Ctrl-C), then, as that worker: -jig worker publish-retry --data ~/.jig/worker +jig worker publish-retry --data ~/trial/worker ``` Use the same `--data` directory and the same `--commit-author-name`, diff --git a/docs/quickstart.md b/docs/quickstart.md index 9763983..fe2f727 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -239,7 +239,7 @@ worker that ran the job: ```sh # stop the worker that ran the job (Ctrl-C), then, as that worker: -jig worker publish-retry --data ~/.jig/worker +./bin/jig worker publish-retry --data ~/.jig/worker ``` Use the same `--data` directory and the same `--commit-author-name`, diff --git a/internal/engine/code_phase.go b/internal/engine/code_phase.go index aa7343b..669acaf 100644 --- a/internal/engine/code_phase.go +++ b/internal/engine/code_phase.go @@ -89,18 +89,21 @@ func runShellCommand( if err := shell.Start(); err != nil { return commandResult{ExitCode: -1, StartError: err.Error()} } - done := make(chan error, 1) - go func() { done <- shell.Wait() }() + // Nothing reaps the shell until its group is recorded: an unreaped + // leader keeps its pid, so the group id cannot be reused by a stranger + // before the kill below — however early the command exits. groupID := int64(shell.Process.Pid) if record != nil { if err := record(groupID, true); err != nil { _ = syscall.Kill(-shell.Process.Pid, syscall.SIGKILL) - <-done + _ = shell.Wait() _ = record(groupID, false) return commandResult{ExitCode: -1, RecordError: err.Error()} } defer func() { _ = record(groupID, false) }() } + done := make(chan error, 1) + go func() { done <- shell.Wait() }() select { case err := <-done: last, overflow := lines.Last() diff --git a/internal/engine/command_record_test.go b/internal/engine/command_record_test.go index f6fc0d6..f883d9e 100644 --- a/internal/engine/command_record_test.go +++ b/internal/engine/command_record_test.go @@ -10,12 +10,12 @@ package engine_test import ( "context" "errors" + "log/slog" "os" "path/filepath" "strconv" "strings" "sync" - "syscall" "testing" "time" @@ -25,13 +25,14 @@ import ( "github.com/StructuPath/jig/internal/worker" ) -// groupRecorder is a RecordProcess fake: the live set per attempt, and an -// optional failure for every active record. +// groupRecorder is a RecordProcess fake: the live set per attempt, an +// optional failure for every active record, and a number of clears to fail. type groupRecorder struct { - mutex sync.Mutex - live map[int64]bool - calls []string - failure error + mutex sync.Mutex + live map[int64]bool + calls []string + failure error + clearFailures int } func (r *groupRecorder) record(attemptID string, groupID int64, active bool) error { @@ -47,6 +48,10 @@ func (r *groupRecorder) record(attemptID string, groupID int64, active bool) err r.live[groupID] = true return r.failure } + if r.clearFailures > 0 { + r.clearFailures-- + return errors.New("manifest clear failed") + } delete(r.live, groupID) return nil } @@ -63,6 +68,14 @@ func (r *groupRecorder) liveCount() int { return len(r.live) } +// callLog is a copy of every record call so far, safe to read while the +// engine may still be recording. +func (r *groupRecorder) callLog() []string { + r.mutex.Lock() + defer r.mutex.Unlock() + return append([]string(nil), r.calls...) +} + // heldCommand is a shell command that writes its own process id — its // group's id, since it leads its group — and then waits for release. func heldCommand(t *testing.T) (command, pidFile, release string) { @@ -71,20 +84,30 @@ func heldCommand(t *testing.T) (command, pidFile, release string) { pidFile = filepath.Join(dir, "pid") release = filepath.Join(dir, "release") command = "echo $$ > '" + pidFile + "'; while [ ! -f '" + release + "' ]; do sleep 0.05; done" - // A failing assertion must not leave the loop running: release it, and - // stop THIS command's own group — never 0, 1, or the test's own. + return command, pidFile, release +} + +// executeHeld runs the attempt in the background. Whatever the test does, +// cleanup releases the held command and joins the execution, so the loop +// ends by its own exit and the engine reaps it: no group id is ever +// signalled from here, so a recycled pid can never be hit. +func executeHeld(t *testing.T, runner *engine.Runner, attempt engine.Attempt, release string) <-chan worker.Outcome { + t.Helper() + outcomes := make(chan worker.Outcome, 1) + finished := make(chan struct{}) + go func() { + defer close(finished) + outcomes <- runner.Execute(context.Background(), attempt) + }() t.Cleanup(func() { _ = os.WriteFile(release, nil, 0o644) - body, err := os.ReadFile(pidFile) - if err != nil { - return - } - if group, err := strconv.Atoi(strings.TrimSpace(string(body))); err == nil && - group > 1 && group != syscall.Getpgrp() { - _ = syscall.Kill(-group, syscall.SIGKILL) + select { + case <-finished: + case <-time.After(30 * time.Second): + t.Error("the held command's attempt did not finish after release") } }) - return command, pidFile, release + return outcomes } func awaitPID(t *testing.T, pidFile string) int64 { @@ -104,6 +127,19 @@ func awaitPID(t *testing.T, pidFile string) int64 { return 0 } +// awaitLive waits, bounded, for the group to be recorded: the shell may write +// its pid a moment before the engine's record call lands. +func awaitLive(t *testing.T, recorder *groupRecorder, group int64) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for !recorder.isLive(group) { + if time.Now().After(deadline) { + t.Fatalf("the running command's group %d was never recorded live (calls %v)", group, recorder.callLog()) + } + time.Sleep(5 * time.Millisecond) + } +} + // Both jig-side command sites record their group live while it runs and // clear it once it has exited. func TestCodeAndGateCommandsAreRecordedLiveWhileTheyRun(t *testing.T) { @@ -126,12 +162,8 @@ func TestCodeAndGateCommandsAreRecordedLiveWhileTheyRun(t *testing.T) { command, pidFile, release := heldCommand(t) snapshot := "name: t\n" + oneWriterRoster + "phases:\n" + phases(command) + "acceptance: [all_phases_passed]\n" - outcomes := make(chan worker.Outcome, 1) - go func() { outcomes <- runner.Execute(context.Background(), testAttempt(snapshot, nil, repo)) }() - pid := awaitPID(t, pidFile) - if !recorder.isLive(pid) { - t.Fatalf("the running command's group %d is not recorded live (calls %v)", pid, recorder.calls) - } + outcomes := executeHeld(t, runner, testAttempt(snapshot, nil, repo), release) + awaitLive(t, recorder, awaitPID(t, pidFile)) if err := os.WriteFile(release, nil, 0o644); err != nil { t.Fatal(err) } @@ -140,7 +172,72 @@ func TestCodeAndGateCommandsAreRecordedLiveWhileTheyRun(t *testing.T) { t.Fatalf("state = %q (%s), want accepted_unpublished", outcome.State, outcome.Error) } if recorder.liveCount() != 0 { - t.Fatalf("groups still recorded live after the command exited: calls %v", recorder.calls) + t.Fatalf("groups still recorded live after the command exited: calls %v", recorder.callLog()) + } + }) + } +} + +// lockedBuffer is a log sink the engine may write from several goroutines. +type lockedBuffer struct { + mutex sync.Mutex + text strings.Builder +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mutex.Lock() + defer b.mutex.Unlock() + return b.text.Write(p) +} + +func (b *lockedBuffer) String() string { + b.mutex.Lock() + defer b.mutex.Unlock() + return b.text.String() +} + +// A clear that fails is tried once more at once. A transient failure is +// absorbed — the group is cleared and nothing is logged — and a persistent +// one stops after that second try, leaves the group recorded, and is logged; +// either way the command already ran, and its outcome stands. +func TestAFailedClearIsRetriedOnceAndNeverChangesTheOutcome(t *testing.T) { + for name, tc := range map[string]struct { + clearFailures int + calls int // record calls: one active, then the clears + stillLive bool + warned bool + }{ + "transient": {clearFailures: 1, calls: 3, stillLive: false, warned: false}, + "persistent": {clearFailures: 5, calls: 3, stillLive: true, warned: true}, + } { + t.Run(name, func(t *testing.T) { + repo := initRepo(t) + recorder := &groupRecorder{live: map[int64]bool{}, clearFailures: tc.clearFailures} + logs := &lockedBuffer{} + fake := enginetest.New() + runner := newTestRunner(t, fake, &recordingSink{}, func(config *engine.Config) { + config.RecordProcess = recorder.record + config.Logger = slog.New(slog.NewTextHandler(logs, nil)) + }) + snapshot := "name: t\n" + oneWriterRoster + `phases: + - name: verify + kind: code + command: "true" +acceptance: [all_phases_passed] +` + outcome := runner.Execute(context.Background(), testAttempt(snapshot, nil, repo)) + if outcome.State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("state = %q (%s), want accepted_unpublished: the command ran and passed", outcome.State, outcome.Error) + } + if calls := recorder.callLog(); len(calls) != tc.calls { + t.Fatalf("record calls = %v, want %d: one active record and at most one retried clear", calls, tc.calls) + } + if live := recorder.liveCount() == 1; live != tc.stillLive { + t.Fatalf("group still recorded = %v, want %v", live, tc.stillLive) + } + warned := strings.Contains(logs.String(), "process_group_record_failed") + if warned != tc.warned { + t.Fatalf("warning logged = %v, want %v\n%s", warned, tc.warned, logs) } }) } @@ -193,7 +290,7 @@ func TestACommandWhoseGroupCannotBeRecordedIsKilledAndFailsAsInfrastructure(t *t t.Fatalf("the unrecorded command kept running and wrote its marker (stat: %v)", err) } if recorder.liveCount() != 0 { - t.Fatalf("the failed record was never cleared: calls %v", recorder.calls) + t.Fatalf("the failed record was never cleared: calls %v", recorder.callLog()) } }) } diff --git a/internal/engine/phase.go b/internal/engine/phase.go index b75d426..feb2911 100644 --- a/internal/engine/phase.go +++ b/internal/engine/phase.go @@ -946,9 +946,13 @@ func (e *execution) runGates( // commandRecorder is the record callback for jig-side commands — a code // phase's command and a tests_pass gate's — into the same manifest set agent -// process groups go to (U4). Nil when nothing records. A failed clear is -// logged exactly as an agent's is; a failed start is returned, and the -// command is killed rather than run unrecorded (runShellCommand). +// process groups go to (U4). Nil when nothing records. A failed start is +// returned, and the command is killed rather than run unrecorded +// (runShellCommand). A failed clear is tried once more at once: a group left +// recorded after its command exited is one a later restart re-examines, and +// a transient write failure should not leave it there. A clear that still +// fails is logged exactly as an agent's is, and the command's own outcome +// stands — it already ran. func (e *execution) commandRecorder() func(groupID int64, active bool) error { if e.runner.config.RecordProcess == nil { return nil @@ -956,6 +960,9 @@ func (e *execution) commandRecorder() func(groupID int64, active bool) error { attemptID := e.attempt.Claim.Attempt.ID return func(groupID int64, active bool) error { err := e.runner.config.RecordProcess(attemptID, groupID, active) + if err != nil && !active { + err = e.runner.config.RecordProcess(attemptID, groupID, active) + } if err != nil { e.runner.config.Logger.Warn("process_group_record_failed", "attempt_id", attemptID, "process_group_id", groupID, "active", active, "error", err)