fix: stop treating a Windows sharing collision as a lost job record

CHANGELOG.md

@@ -6,6 +6,32 @@ All notable packetcode changes are recorded here. The project is pre-1.0; `Unrel
 
 ### Fixed
 
+- **Job records could silently disappear on Windows.** Windows opens deny by
+  default, and Go's `os.ReadFile` asks for `FILE_SHARE_READ|WRITE` but not
+  `FILE_SHARE_DELETE`. So while any reader holds a job record open, the rename
+  that publishes a new version of it fails with `ERROR_ACCESS_DENIED`, and
+  while that rename is in flight a reader fails with
+  `ERROR_SHARING_VIOLATION`. Both are "try again in a moment", which is what
+  POSIX does implicitly; both were being treated as permanent. Measured on one
+  contended path: 809 of 2000 renames and 857 concurrent reads failed. The
+  consequences were real — a terminal job state whose write failed is
+  discarded silently by every `_ = m.savePersistedSnapshot...` call site, and a
+  record whose read failed is reported as *malformed* by `decodeRecordFile` and
+  dropped from the reload entirely. `atomicfile` now waits such a collision out
+  (ten attempts, 10ms apart, Windows only; the loops compile to a single pass
+  everywhere else) and exposes `atomicfile.ReadFile` for the read half, which
+  the job record readers use. A regression test contends a reader and a writer
+  on one path and fails without the retry.
+- `TestResubmit_SpawnsNewJobAndLinksBothWays` was flaky on Windows CI as a
+  result of the above, plus a mistake of its own: it waited for `Manager.Get`
+  to report a terminal state and then read the record off disk, but
+  `markTerminalCause` flips the in-memory state under the manager lock and
+  persists only after releasing it. Reading inside that window found the
+  successor still `running`, which sent the loader down its reconcile-and-
+  rewrite path against a file the manager was writing at the same instant. It
+  now waits for the record itself, and asserts the loader reported nothing
+  unreadable — the discarded `unreadable` return is why the failure only ever
+  said "map does not contain <id>".
 - `TestRunUserPromptSubmit_CollectsStdout` failed on `test (windows-latest)`
   about two runs in three and never on a developer machine. The cause was
   measured rather than guessed: on four GitHub `windows-latest` runners the

internal/atomicfile/atomicfile.go

@@ -18,6 +18,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"time"
 )
 
 // Write writes data to path via a temp file in the same directory, fsynced
@@ -55,14 +56,61 @@ func Write(path string, data []byte, perm os.FileMode, tmpPattern string) error
 		_ = os.Remove(tmpPath)
 		return fmt.Errorf("close temp: %w", err)
 	}
-	if err := os.Rename(tmpPath, path); err != nil {
+	if err := renameRetrying(tmpPath, path); err != nil {
 		_ = os.Remove(tmpPath)
 		return fmt.Errorf("rename: %w", err)
 	}
 	syncDir(dir)
 	return nil
 }
 
+// shareRetries and shareRetryDelay bound how long an operation waits out a
+// Windows sharing collision: ten attempts, ten milliseconds apart, so about a
+// tenth of a second in the worst case and no wait at all in the common one.
+//
+// Bounded on purpose. A file genuinely held open by another program -- an
+// editor, a virus scanner with a long lease -- must still fail and say so; the
+// retry is for the millisecond-scale window in which two of our own goroutines
+// touch the same record, which is the only case observed here.
+const (
+	shareRetries    = 10
+	shareRetryDelay = 10 * time.Millisecond
+)
+
+// renameRetrying is os.Rename that waits out a transient Windows sharing
+// collision instead of reporting one as a failed write.
+func renameRetrying(from, to string) error {
+	var err error
+	for attempt := 0; attempt < shareRetries; attempt++ {
+		if err = os.Rename(from, to); err == nil || !isShareViolation(err) {
+			return err
+		}
+		time.Sleep(shareRetryDelay)
+	}
+	return err
+}
+
+// ReadFile is os.ReadFile that waits out a transient Windows sharing collision.
+//
+// It is the read half of the same problem Write solves: a reader that happens
+// to open a file during the instant a rename is replacing it gets
+// ERROR_SHARING_VIOLATION, which callers cannot distinguish from a corrupt or
+// missing record and so tend to report as one. A file that does not exist still
+// returns os.ErrNotExist on the first attempt, without waiting.
+func ReadFile(path string) ([]byte, error) {
+	var (
+		data []byte
+		err  error
+	)
+	for attempt := 0; attempt < shareRetries; attempt++ {
+		if data, err = os.ReadFile(path); err == nil || !isShareViolation(err) {
+			return data, err
+		}
+		time.Sleep(shareRetryDelay)
+	}
+	return data, err
+}
+
 // syncDir flushes the directory entry so the rename itself survives a crash,
 // and not merely the bytes it points at.
 //

internal/atomicfile/share_other.go

@@ -0,0 +1,11 @@
+//go:build !windows
+
+package atomicfile
+
+// isShareViolation is always false away from Windows.
+//
+// POSIX renames succeed with the destination open, and a reader holding an
+// unlinked inode keeps reading it, so there is no transient state to retry
+// through. The retry loops below therefore run exactly once here: same syscall
+// count, same behaviour, no sleep.
+func isShareViolation(error) bool { return false }

internal/atomicfile/share_race_test.go

@@ -0,0 +1,82 @@
+package atomicfile
+
+import (
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+// A reader and a writer contending on one path must both succeed.
+//
+// Without the retry this failed on Windows in roughly a quarter of attempts:
+// os.Rename onto a path a reader has open returns ERROR_ACCESS_DENIED, and a
+// read landing inside a rename returns ERROR_SHARING_VIOLATION. Neither means
+// the file is bad, and callers that treated them as permanent lost records.
+//
+// On POSIX this asserts the same invariant, where it has always held.
+func TestWriteAndReadContendOnOnePathWithoutFailing(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "record.json")
+	if err := os.WriteFile(path, []byte(`{"state":"running"}`), 0o600); err != nil {
+		t.Fatalf("seed: %v", err)
+	}
+
+	const writes = 300
+	var readErr, writeErr atomic.Value
+	var reads int64
+
+	done := make(chan struct{})
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-done:
+				return
+			default:
+				if _, err := ReadFile(path); err != nil {
+					readErr.Store(err)
+					return
+				}
+				atomic.AddInt64(&reads, 1)
+			}
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		defer close(done)
+		for i := 0; i < writes; i++ {
+			if err := Write(path, []byte(`{"state":"completed"}`), 0o600, ".record.*.json.tmp"); err != nil {
+				writeErr.Store(err)
+				return
+			}
+		}
+	}()
+	wg.Wait()
+
+	if err, ok := writeErr.Load().(error); ok && err != nil {
+		t.Fatalf("a write lost a race with a concurrent reader: %v", err)
+	}
+	if err, ok := readErr.Load().(error); ok && err != nil {
+		t.Fatalf("a read lost a race with a concurrent writer: %v", err)
+	}
+	if got := atomic.LoadInt64(&reads); got == 0 {
+		t.Fatal("the reader never completed a read, so nothing was contended")
+	}
+	// No temp files may be left behind by a retried rename.
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatalf("read dir: %v", err)
+	}
+	for _, e := range entries {
+		if e.Name() != "record.json" {
+			t.Errorf("leftover file after contended writes: %s", e.Name())
+		}
+	}
+}

internal/atomicfile/share_windows.go

@@ -0,0 +1,37 @@
+//go:build windows
+
+package atomicfile
+
+import (
+	"errors"
+	"syscall"
+
+	"golang.org/x/sys/windows"
+)
+
+// isShareViolation reports whether err is Windows refusing an operation only
+// because someone else has the file open at this instant.
+//
+// Windows opens deny by default. Go's os.ReadFile asks for FILE_SHARE_READ and
+// FILE_SHARE_WRITE but not FILE_SHARE_DELETE, so while any reader holds a
+// handle, a rename onto that path fails with ERROR_ACCESS_DENIED -- and while a
+// rename is replacing the file, a reader fails with ERROR_SHARING_VIOLATION.
+// Neither says anything is wrong with the file or the caller; both mean "try
+// again in a moment", which is what POSIX does implicitly by allowing the
+// rename to proceed under an open handle.
+//
+// Measured on this repository's own job records: with one reader and one writer
+// contending on a single path, 809 of 2000 renames failed with errno 5 and 857
+// reads failed with errno 32. Treating those as permanent is what let a job
+// record vanish and be reported as malformed.
+func isShareViolation(err error) bool {
+	if err == nil {
+		return false
+	}
+	var errno syscall.Errno
+	if !errors.As(err, &errno) {
+		return false
+	}
+	return errno == syscall.Errno(windows.ERROR_ACCESS_DENIED) ||
+		errno == syscall.Errno(windows.ERROR_SHARING_VIOLATION)
+}

internal/jobs/persistence.go

@@ -317,7 +317,7 @@ func savePersistedSnapshot(jobsDir string, p persistedJob) error {
 }
 
 func readPersistedJob(path string) (persistedJob, bool) {
-	data, err := os.ReadFile(path)
+	data, err := atomicfile.ReadFile(path)
 	if err != nil {
 		return persistedJob{}, false
 	}
@@ -332,7 +332,11 @@ func readPersistedJob(path string) (persistedJob, bool) {
 // Both the loader and the read-only inspector go through it so a record that
 // one of them calls unreadable is never quietly accepted by the other.
 func decodeRecordFile(path string) (persistedJob, State, *UnreadableRecord) {
-	data, err := os.ReadFile(path)
+	// atomicfile.ReadFile, not os.ReadFile: on Windows a read that lands in the
+	// instant a rename is replacing the record fails with a sharing violation,
+	// and reporting that as an unreadable record is how a perfectly good job
+	// silently disappeared from a reload.
+	data, err := atomicfile.ReadFile(path)
 	if err != nil {
 		return persistedJob{}, StateFailed, &UnreadableRecord{Path: path, Reason: err.Error()}
 	}

internal/jobs/resubmit_test.go

@@ -1,6 +1,7 @@
 package jobs
 
 import (
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -110,12 +111,29 @@ func TestResubmit_SpawnsNewJobAndLinksBothWays(t *testing.T) {
 	assert.Equal(t, snap.ID, before.ResubmittedAs)
 
 	// Both links must be durable across a reload.
-	waitFor(t, 5*time.Second, "successor terminal", func() bool {
-		s, ok := mgr.Get(snap.ID)
-		return ok && s.State.IsTerminal()
+	//
+	// Wait for the successor's *record*, not for mgr.Get to report a terminal
+	// state. markTerminalCause flips the in-memory state under the manager
+	// lock and only persists after releasing it, so "terminal in memory" does
+	// not yet mean "terminal on disk". Reading during that window used to find
+	// the record still Running, which sent the loader down its reconcile-and-
+	// rewrite path against a file the manager was writing at the same moment --
+	// and on Windows one of those two collides and the record is dropped.
+	//
+	// readPersistedJob is the right instrument for the wait because it only
+	// reads. Polling loadPersistedJobs would rewrite what it is waiting on.
+	successorRecord := filepath.Join(jobsDir, snap.ID+".json")
+	waitFor(t, 5*time.Second, "successor terminal on disk", func() bool {
+		p, ok := readPersistedJob(successorRecord)
+		return ok && parseState(p.State).IsTerminal()
 	})
-	reloaded, _, _, lerr := loadPersistedJobs(jobsDir, "")
+
+	reloaded, _, unreadable, lerr := loadPersistedJobs(jobsDir, "")
 	require.NoError(t, lerr)
+	// Asserted rather than discarded: a record the loader rejects is dropped
+	// from `reloaded`, so without this the failure is "map does not contain
+	// <id>" and says nothing about why. It cost an afternoon once.
+	require.Empty(t, unreadable, "every record written by this test must load back")
 	byID := map[string]*Job{}
 	for _, j := range reloaded {
 		byID[j.ID] = j