Commit fe39804

Eric Bower  ·  2026-08-11 11:10:59 -0400 EDT
parent 6702fca 
fix: more sync issues with the `pici` local runner
2 files changed,  +143, -19
M main.go
M main_test.go
M main.go
+67, -15
......@@ -448,9 +448,9 @@ func (w *WorkspaceRsync) Setup() error {
448448 w.Cfg.KeyLocation,
449449 w.Cfg.CertificateLocation,
450450 )
451- cmd = exec.Command("rsync", "-e", sshcmd, "-rv", "--exclude=/.git", "--exclude=/.jj", w.Source+"/", w.Dest+"/")
451+ cmd = exec.Command("rsync", "-e", sshcmd, "-rv", "--exclude=/.git", "--exclude=/.jj", "--filter=:- .gitignore", w.Source+"/", w.Dest+"/")
452452 } else {
453- cmd = exec.Command("rsync", "-rv", "--exclude=/.git", "--exclude=/.jj", w.Source+"/", w.Dest+"/")
453+ cmd = exec.Command("rsync", "-rv", "--exclude=/.git", "--exclude=/.jj", "--filter=:- .gitignore", w.Source+"/", w.Dest+"/")
454454 }
455455 if err := runCmd(cmd, log); err != nil {
456456 return err
......@@ -990,8 +990,11 @@ func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID, eventType string) er
990990 }
991991 liveLines = lines
992992
993- // Check if all sessions are done
994- if allCompleted(jobSessions) {
993+ // Stage local artifacts on tick so progress and HTML reports update live
994+ _ = stageLocalArtifacts(cfg, log, name, jobID, eventType)
995+
996+ // Check if all sessions are done and the runner session has finished
997+ if isJobComplete(jobSessions) {
995998 break
996999 }
9971000 }
......@@ -1308,7 +1311,7 @@ func monitorTick(cfg *Cfg, log *slog.Logger, output io.Writer, jobStates map[str
13081311 // Compute job-level timing from session timestamps
13091312 startedAt, endedAt, duration := computeJobTiming(group)
13101313
1311- if allCompleted(group) {
1314+ if isJobComplete(group) {
13121315 // Check sentinel — publish terminal status exactly once
13131316 sentinel := filepath.Join(cfg.ArtifactDir, name, jobID, "artifacts", "published.json")
13141317 log.Debug("checking completion", "all_completed", true, "sentinel", sentinel)
......@@ -1772,6 +1775,26 @@ func allCompleted(sessions []SessionInfo) bool {
17721775 return true
17731776 }
17741777
1778+// isJobComplete returns true if all sessions for a job have completed
1779+// AND a runner session exists and has completed. This ensures monitor and wait
1780+// do not mark a job complete while pico.sh (the runner) is still executing and
1781+// potentially spawning new tasks.
1782+func isJobComplete(sessions []SessionInfo) bool {
1783+ if len(sessions) == 0 {
1784+ return false
1785+ }
1786+ hasRunner := false
1787+ for _, s := range sessions {
1788+ if s.Ended == "" {
1789+ return false
1790+ }
1791+ if strings.HasSuffix(s.Name, ".runner") {
1792+ hasRunner = true
1793+ }
1794+ }
1795+ return hasRunner
1796+}
1797+
17751798 func runCmd(cmd *exec.Cmd, log *slog.Logger) error {
17761799 stdout, err := cmd.StdoutPipe()
17771800 if err != nil {
......@@ -1918,8 +1941,25 @@ func cancelJobSessions(prefix string) {
19181941 }
19191942
19201943 // findSessionsForGC identifies finished sessions or sessions older than 3 hours matching given prefixes.
1944+// It checks whether a session's job has an active runner session (a .runner session with Ended == "");
1945+// if so, finished child sessions for that job are NOT cleaned up yet to ensure no new tasks spawn
1946+// in pico.sh before cleaning up tasks.
19211947 func findSessionsForGC(sessions []SessionInfo, prefixes []string, now time.Time) []string {
19221948 cutoff := now.Add(-3 * time.Hour).Unix()
1949+
1950+ // Track which job prefixes have ANY active (running) session.
1951+ // As long as any session for a job is still running (Ended == ""),
1952+ // the job is active and none of its sessions should be garbage collected.
1953+ activeJobPrefixes := make(map[string]bool)
1954+ for _, s := range sessions {
1955+ if s.Ended == "" {
1956+ prefix := extractJobPrefix(s.Name)
1957+ if prefix != "" {
1958+ activeJobPrefixes[prefix] = true
1959+ }
1960+ }
1961+ }
1962+
19231963 var toKill []string
19241964 for _, s := range sessions {
19251965 matched := false
......@@ -1933,6 +1973,22 @@ func findSessionsForGC(sessions []SessionInfo, prefixes []string, now time.Time)
19331973 continue
19341974 }
19351975
1976+ // If the session belongs to a job with an active runner, do not kill it
1977+ // unless it has expired (> 3 hours old).
1978+ jobPrefix := extractJobPrefix(s.Name)
1979+ if jobPrefix != "" && activeJobPrefixes[jobPrefix] {
1980+ if s.Created != "" {
1981+ var created int64
1982+ if _, err := fmt.Sscanf(s.Created, "%d", &created); err != nil {
1983+ continue
1984+ }
1985+ if created < cutoff {
1986+ toKill = append(toKill, s.Name)
1987+ }
1988+ }
1989+ continue
1990+ }
1991+
19361992 if s.Ended != "" {
19371993 toKill = append(toKill, s.Name)
19381994 continue
......@@ -2322,7 +2378,7 @@ func runLocal(cfg *Cfg, dest string) error {
23222378 }
23232379
23242380 // Generate and stage full HTML/txt artifacts and index
2325- if err := stageLocalArtifacts(cfg, log, eventData.Name, jobID); err != nil {
2381+ if err := stageLocalArtifacts(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
23262382 log.Error("stage local artifacts", "err", err)
23272383 }
23282384
......@@ -2341,22 +2397,18 @@ func runLocal(cfg *Cfg, dest string) error {
23412397 return nil
23422398 }
23432399
2344-func stageLocalArtifacts(cfg *Cfg, log *slog.Logger, repoName, jobID string) error {
2400+func stageLocalArtifacts(cfg *Cfg, log *slog.Logger, repoName, jobID, eventType string) error {
2401+ domain := getDomain(eventType)
2402+ prefix := fmt.Sprintf("%s.%s.%s.", domain, repoName, jobID)
2403+
23452404 listOutput, err := exec.Command("zmx", "list").CombinedOutput()
23462405 if err != nil {
23472406 return fmt.Errorf("zmx list: %w", err)
23482407 }
23492408 sessions := parseZMXList(string(listOutput))
2350- var localSessions []SessionInfo
2351- for _, s := range sessions {
2352- if strings.HasPrefix(s.Name, "local.") {
2353- localSessions = append(localSessions, s)
2354- }
2355- }
2356- prefix := fmt.Sprintf("local.%s.%s.", repoName, jobID)
23572409
23582410 var jobSessions []SessionInfo
2359- for _, s := range localSessions {
2411+ for _, s := range sessions {
23602412 if strings.HasPrefix(s.Name, prefix) {
23612413 s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
23622414 jobSessions = append(jobSessions, s)
M main_test.go
+76, -4
......@@ -519,6 +519,60 @@ func TestAllCompleted(t *testing.T) {
519519 }
520520 }
521521
522+func TestIsJobComplete(t *testing.T) {
523+ tests := []struct {
524+ name string
525+ sessions []SessionInfo
526+ want bool
527+ }{
528+ {
529+ name: "all completed with runner",
530+ sessions: []SessionInfo{
531+ {Name: "ci.repo.abc.runner", Ended: "100"},
532+ {Name: "ci.repo.abc.step.fmt", Ended: "90"},
533+ },
534+ want: true,
535+ },
536+ {
537+ name: "runner active, step completed",
538+ sessions: []SessionInfo{
539+ {Name: "ci.repo.abc.runner", Ended: ""},
540+ {Name: "ci.repo.abc.step.fmt", Ended: "90"},
541+ },
542+ want: false,
543+ },
544+ {
545+ name: "no runner session present",
546+ sessions: []SessionInfo{
547+ {Name: "ci.repo.abc.step.fmt", Ended: "90"},
548+ },
549+ want: false,
550+ },
551+ {
552+ name: "runner completed, step active",
553+ sessions: []SessionInfo{
554+ {Name: "ci.repo.abc.runner", Ended: "100"},
555+ {Name: "ci.repo.abc.step.fmt", Ended: ""},
556+ },
557+ want: false,
558+ },
559+ {
560+ name: "empty sessions",
561+ sessions: []SessionInfo{},
562+ want: false,
563+ },
564+ }
565+
566+ for _, tt := range tests {
567+ t.Run(tt.name, func(t *testing.T) {
568+ got := isJobComplete(tt.sessions)
569+ if got != tt.want {
570+ t.Errorf("isJobComplete() = %v, want %v", got, tt.want)
571+ }
572+ })
573+ }
574+}
575+
522576 func TestParseZMXList(t *testing.T) {
523577 output := `name=ci-lint pid=1064464 clients=0 created=1777519944 start_dir=/home/erock/dev/pico ended=1777519986 exit_code=0
524578 name=ci-tests pid=1064472 clients=0 created=1777519944 start_dir=/home/erock/dev/pico ended=1777519958 exit_code=2
......@@ -645,17 +699,26 @@ func TestFindSessionsForGC(t *testing.T) {
645699
646700 sessions := []SessionInfo{
647701 {Name: "ci.repo.active.runner", Created: twoHoursAgo, Ended: ""},
702+ {Name: "ci.repo.active.step.fmt", Created: twoHoursAgo, Ended: "1786289500"},
648703 {Name: "ci.repo.finished.runner", Created: twoHoursAgo, Ended: "1786289600"},
704+ {Name: "ci.repo.finished.step.lint", Created: twoHoursAgo, Ended: "1786289550"},
649705 {Name: "ci.repo.expired.runner", Created: fourHoursAgo, Ended: ""},
650706 {Name: "other.session", Created: fourHoursAgo, Ended: ""},
651707 }
652708
653709 toKill := findSessionsForGC(sessions, []string{"ci.", "local."}, now)
654- if len(toKill) != 2 {
655- t.Fatalf("expected 2 sessions for GC, got %d: %v", len(toKill), toKill)
710+ if len(toKill) != 3 {
711+ t.Fatalf("expected 3 sessions for GC, got %d: %v", len(toKill), toKill)
656712 }
657- if toKill[0] != "ci.repo.finished.runner" || toKill[1] != "ci.repo.expired.runner" {
658- t.Errorf("unexpected sessions for GC: %v", toKill)
713+ expected := map[string]bool{
714+ "ci.repo.finished.runner": true,
715+ "ci.repo.finished.step.lint": true,
716+ "ci.repo.expired.runner": true,
717+ }
718+ for _, k := range toKill {
719+ if !expected[k] {
720+ t.Errorf("unexpected session in GC kill list: %s", k)
721+ }
659722 }
660723 }
661724
......@@ -858,6 +921,12 @@ func TestWorkspaceRsync_ExcludesGitAndJJ(t *testing.T) {
858921 if err := os.WriteFile(filepath.Join(srcDir, "hello.txt"), []byte("world"), 0644); err != nil {
859922 t.Fatalf("failed to write hello.txt: %v", err)
860923 }
924+ if err := os.WriteFile(filepath.Join(srcDir, ".gitignore"), []byte("ignored.txt\n"), 0644); err != nil {
925+ t.Fatalf("failed to write .gitignore: %v", err)
926+ }
927+ if err := os.WriteFile(filepath.Join(srcDir, "ignored.txt"), []byte("secret"), 0644); err != nil {
928+ t.Fatalf("failed to write ignored.txt: %v", err)
929+ }
861930 if err := os.MkdirAll(filepath.Join(srcDir, ".git"), 0755); err != nil {
862931 t.Fatalf("failed to mkdir .git: %v", err)
863932 }
......@@ -888,6 +957,9 @@ func TestWorkspaceRsync_ExcludesGitAndJJ(t *testing.T) {
888957 if _, err := os.Stat(filepath.Join(destDir, "hello.txt")); os.IsNotExist(err) {
889958 t.Error("expected hello.txt to exist in dest workspace")
890959 }
960+ if _, err := os.Stat(filepath.Join(destDir, "ignored.txt")); !os.IsNotExist(err) {
961+ t.Error("expected ignored.txt to be excluded by .gitignore from dest workspace")
962+ }
891963 if _, err := os.Stat(filepath.Join(destDir, ".git")); !os.IsNotExist(err) {
892964 t.Error("expected .git to be excluded from dest workspace")
893965 }