Fix write-concurrency race: per-file write protocol + unique temps
The review-identified race: saves are async (owner snapshots content,
worker pool writes), nothing serialized per file, and every write of a
file used the SAME deterministic temp ('.<name>.tmp'). Two overlapping
writes (autosave x autosave, retry x autosave, or the synchronous
FlushAll on GoToBrowser/Shutdown x a worker write) interleaved on the
shared temp and could rename a byte-mixture into place; even without
interleaving, last-rename-wins could promote a STALE snapshot.
Owner-side protocol (logic.go, requestSave + result handler):
- at most one write in flight per file (writeInFlight maps filename ->
the file version whose content the in-flight write carries);
- a save requested while one is in flight is deferred (savePending) and
re-issued by the write's result handler with a FRESH snapshot, so
'last rename wins' coincides with 'newest snapshot wins';
- on success the SNAPSHOT version (not the current one) is recorded as
written, so an edit that arrived during the write leaves the file
dirty and triggers the re-issue;
- FlushAll (GoToBrowser, Shutdown) defers via the same protocol instead
of writing concurrently on the shared temp;
- shutdown drain: on done the owner waits (bounded 5 s) for in-flight
writes and armed retries to settle before exiting, so the post-exit
synchronous FlushAll and workerPool.Stop cannot race a straggling
worker write;
- retry timer now sends a non-blocking token (no timer-goroutine stall
on a full channel); emitFrame no longer blocks on a slow/gone main
(frames are snapshots; the next emission wins) - also required so the
drain can never deadlock on frame delivery.
Mechanism (real/filesystem.go):
- WriteFileAtomic uses a unique per-call temp ('.<name>.tmp.<pid>.<seq>'),
making same-file staging-file interleaving structurally impossible even
if the serialization regressed (defense in depth);
- each successful write best-effort removes stale temps of the same file
(crash leftovers, plus the legacy deterministic name for upgraded
installs); a failed write removes its own temp.
Tests:
- write_serialization_test.go (e2e): a counting FS wrapper proves the
peak concurrent same-file saves is 1 across two deliberately
overlapping autosaves (2 s saves; the second edit lands inside the
first save's window and its token is deferred, then re-issued with the
newer content), and that a Flush during an in-flight save adds no
concurrent writer and the newest snapshot still wins. Mutation-verified:
disabling the deferral fails it with peak = 2. (The pool's WriteFileTask
calls FS.WriteFile, not WriteFileAtomic - the real FS is atomic only
because WriteFile delegates to WriteFileAtomic; the wrapper mirrors that
delegation or the overlap window does not exist.)
- filesystem_test.go: stale-temp test updated to the new pattern, also
covering the legacy name and asserting a different file's temp is
untouched.
- real_file_fuzz_test.go: stray-temp check matches both patterns.
Docs: architecture.md 6.5 rewritten (protocol invariants), spec.md
autosave line, development_plan.md v11 + Phase 12.
On-device smoke: open, type, autosave lands exact content on disk, no
temp files left, clean relaunch. Full suite green under -race.
Residuals (documented): no fsync before rename (power-loss window only);
external-change detection absent; a drain-deadline exit with a straggling
write can only lose freshness (unique temps keep every rename a complete
snapshot).
This commit is contained in:
parent
5205906de4
commit
6c6a0c1a27
|
|
@ -369,15 +369,42 @@ Only the visible byte range is shaped and drawn each frame:
|
|||
`deleteSurroundingText` (backspace/autocorrect replacement), and composition
|
||||
all arrive through this one path.
|
||||
|
||||
### 6.5 Autosave
|
||||
### 6.5 Autosave and the per-file write protocol
|
||||
|
||||
- Any edit calls `markDirty()`: a 1 s debounce timer; each keystroke restarts
|
||||
it. On expiry the timer goroutine sends a token on `autosaveChan`; the owner
|
||||
reconstructs the full content from the chunked buffer and dispatches a
|
||||
`WriteFile` task.
|
||||
snapshots the full content from the chunked buffer and requests a save.
|
||||
- **Per-file write protocol (corruption guard).** The worker pool is shared
|
||||
and the on-disk staging path is per-file, so two concurrent writes of the
|
||||
same file would race on the staging file. The owner therefore guarantees:
|
||||
- at most **one write in flight per file** (`writeInFlight`, keyed by
|
||||
filename, recording the file version whose content the write carries);
|
||||
- a save requested while one is in flight is **deferred** (`savePending`)
|
||||
and re-issued by the write's result handler — so the rename that lands
|
||||
last always carries the **newest** content ("latest state wins");
|
||||
- on success the recorded written version is the **snapshot's** version,
|
||||
so any edit that arrived during the write leaves the file dirty and
|
||||
triggers the re-issue.
|
||||
- `FlushAll` (page switch, shutdown) obeys the same protocol: if a write is
|
||||
in flight it defers instead of writing concurrently.
|
||||
- **Shutdown drain:** on the `done` signal the owner waits (bounded, 5 s) for
|
||||
in-flight writes and armed retries to settle before exiting, so the
|
||||
post-exit `FlushAll` and `workerPool.Stop` cannot race a straggling worker
|
||||
write.
|
||||
- **Staging file:** `WriteFileAtomic` uses a **unique per-call temp name**
|
||||
("`.<name>.tmp.<pid>.<seq>`") in the target directory and renames it into
|
||||
place, so readers and crash recovery only ever see a complete file. Each
|
||||
successful write also best-effort removes stale temps of the same file
|
||||
(crash leftovers, plus the legacy deterministic "`.<name>.tmp`" name).
|
||||
Unique temp names make same-file interleaving structurally impossible even
|
||||
if the serialization regressed.
|
||||
- Write failures are tracked per file (`writeFailed`, `retryAttempts`) and
|
||||
retried via `retryChan`. There is no save button; autosave is the only
|
||||
persistence.
|
||||
retried via `retryChan` (exponential backoff 1 s … 30 s; the timer sends a
|
||||
non-blocking token and the owner re-snapshots at fire time). There is no
|
||||
save button; autosave is the only persistence.
|
||||
- Known residual (out of scope): no `fsync` before the rename — a **power
|
||||
loss** inside the rename window can lose the last save (process death
|
||||
cannot: the page cache survives).
|
||||
|
||||
### 6.6 Opening a file
|
||||
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
# Development Plan: reach a lean, usable Android text editor
|
||||
|
||||
Status: v10, 2026-08-17 (Phases 0–3 + doc reorg + Phase 6 scroll-perf/clamping
|
||||
Status: v11, 2026-08-17 (Phases 0–3 + doc reorg + Phase 6 scroll-perf/clamping
|
||||
verification + tap-to-position-cursor fix + selection + real-file e2e +
|
||||
Android arrow/shift workaround + touch selection + scroll-offset tap proof &
|
||||
float32 decomposition fix + font-scale (user font-size setting) support).
|
||||
float32 decomposition fix + font-scale (user font-size setting) support +
|
||||
data-corruption test suite & write-concurrency fix).
|
||||
Written
|
||||
against the **live** repo `/home/gmp/pad`. v1 (the widget-rebuild plan) is
|
||||
superseded — see §12 for why. Doc reorganization (2026-08-16): the over-detailed
|
||||
|
|
@ -497,6 +498,73 @@ configurations, or only on this AVD's density?
|
|||
marker → read the file off the device) is the ground truth and needs no
|
||||
profiler.
|
||||
|
||||
### Phase 12 — data-corruption test suite + write-concurrency fix — DONE (2026-08-17)
|
||||
|
||||
Two questions: (a) how do we *prove* the edit→persist path cannot corrupt
|
||||
file data; (b) fix the write-concurrency race the review identified (user
|
||||
deprioritized it, then asked for it to be implemented after the test suite).
|
||||
|
||||
**Test suite (5 layers, all differential/oracle-based):**
|
||||
- `chunked_buffer_fuzz_test.go` — every random Insert/Delete/replace mirrored
|
||||
on a plain `[]byte` shadow model (chunk sizes 1…64 KB, 500–2000 ops each);
|
||||
after every op: FileLen, FullContent, Content probes, chunk-size invariant.
|
||||
Rune-aligned variant adds UTF-8 validity + an independent RuneIndexToByte
|
||||
oracle.
|
||||
- `line_index_fuzz_test.go` — incremental LineIndex updates vs a
|
||||
full-recomputation oracle (mixed / no-newline / single-line / CRLF /
|
||||
trailing-newline shapes), plus a trailing-empty-line structural test.
|
||||
- `state_api_fuzz_test.go` — the production edit entry points
|
||||
(HandleInsert/Backspace/Delete/ReplaceRange incl. selection variants);
|
||||
after every op: content, UTF-8, chunk invariant, line index, exact cursor.
|
||||
- `real_file_fuzz_test.go` (e2e) — random edit sequences incl. window-relative
|
||||
IME replaces against the real FS; per-batch IME-window-consistency checks,
|
||||
forced-flush disk byte-compare every batch, then a second Logic instance
|
||||
(restart simulation) must reload byte-identical content; no stray temps.
|
||||
- `filesystem_test.go` (real) — the atomicity contract: exact round trips, a
|
||||
concurrent reader never sees a torn file across 150 alternating 2 MB
|
||||
writes, a failed write leaves the original byte-identical, stale temps are
|
||||
consumed.
|
||||
|
||||
**What the testing found (fixed in `5205906`):** `Insert` halved an oversized
|
||||
spliced result once, so a large paste into a non-empty buffer left chunks up
|
||||
to ~P/2 (8× the documented 2× bound at 1 MB/64 KB). `Insert` now re-chunks
|
||||
the oversized result into pieces ≤ chunkSize, so "no chunk exceeds 2× target
|
||||
after any edit" holds universally. Mutation-tested: dropping one byte in
|
||||
Insert and one entry in UpdateLineIndexAfterInsert are both caught in ms.
|
||||
|
||||
**Write-concurrency fix (this round):**
|
||||
- **Protocol (owner-side, per file):** at most one write in flight per file
|
||||
(`writeInFlight[f]` = snapshot version); a save requested during a write is
|
||||
deferred (`savePending`) and re-issued by the result handler; on success the
|
||||
*snapshot's* version is recorded as written, so edits that landed during the
|
||||
write trigger the re-issue — "last rename wins" coincides with "newest
|
||||
snapshot wins". `FlushAll` (GoToBrowser/Shutdown) obeys the same protocol
|
||||
instead of racing a worker write on the same temp file.
|
||||
- **Mechanism (real FS):** `WriteFileAtomic` now uses a unique per-call temp
|
||||
("`.<name>.tmp.<pid>.<seq>`"), so same-file interleaving is structurally
|
||||
impossible even if the serialization regressed; each successful write also
|
||||
best-effort removes stale temps of the same file (crash leftovers + the
|
||||
legacy deterministic name).
|
||||
- **Shutdown drain:** on `done`, the owner waits (bounded 5 s) for in-flight
|
||||
writes and armed retries to settle before exiting, so the post-exit
|
||||
synchronous `FlushAll` and `workerPool.Stop` cannot race a straggling
|
||||
worker write. The retry timer now sends a non-blocking token (no timer-
|
||||
goroutine stall on a full channel), and `emitFrame` no longer blocks on a
|
||||
slow/gone main (dropped frames are snapshots; the next emission wins).
|
||||
- **Proof:** `write_serialization_test.go` — a counting FS wrapper proves the
|
||||
peak concurrent same-file saves is 1 across two deliberately overlapping
|
||||
autosaves (2 s saves, edit inside the first save's window), and that a
|
||||
Flush during an in-flight save neither adds a concurrent writer nor lets a
|
||||
stale snapshot win. The test was mutation-verified: disabling the deferral
|
||||
makes it fail with peak = 2. (Debugging note: the pool's `WriteFileTask`
|
||||
calls `FS.WriteFile`, not `WriteFileAtomic` — the real FS is atomic only
|
||||
because `WriteFile` delegates to `WriteFileAtomic`; the wrapper mirrors
|
||||
that delegation or the timing window doesn't exist.)
|
||||
- **Residuals (documented, out of scope):** no `fsync` before rename (power-
|
||||
loss window only); external-change detection absent; a drain-deadline exit
|
||||
with a straggling write can only lose freshness (unique temps keep every
|
||||
rename a complete snapshot).
|
||||
|
||||
## 6. File-size decision (re-framed)
|
||||
|
||||
v1 framed this as "accept a limit vs build a windowed editor." The live repo
|
||||
|
|
|
|||
|
|
@ -72,8 +72,12 @@ elsewhere.
|
|||
is tracked by the app, since Gio's Android bridge drops modifier keys —
|
||||
architecture.md §2.1.)
|
||||
- **Autosave:** every edit restarts a 1 s debounce; on expiry the full content
|
||||
is written to disk by a worker. Failed writes are retried. This is the only
|
||||
persistence mechanism.
|
||||
is written to disk by a worker. At most one write per file is in flight at
|
||||
any time; saves requested during a write are deferred and re-issued with
|
||||
the newer content when it completes ("latest state wins"). Writes stage to
|
||||
a unique per-call temp file and rename into place, so a crash or a
|
||||
concurrent reader never observes a partial file. Failed writes are retried.
|
||||
This is the only persistence mechanism.
|
||||
|
||||
### 2.3 Large files (measured)
|
||||
|
||||
|
|
|
|||
|
|
@ -73,13 +73,23 @@ type Logic struct {
|
|||
retryChan chan string // auto-save retries
|
||||
autosaveChan chan struct{} // auto-save debounce ticks (timer -> owner)
|
||||
inspectChan chan *inspectReq
|
||||
workerPool *pool.WorkerPool
|
||||
mockFS pool.FileSystem
|
||||
done chan struct{}
|
||||
exitWg sync.WaitGroup
|
||||
saveTimer *time.Timer // auto-save debounce timer; non-nil while pending
|
||||
lastEmit time.Time // time of the last frame emission (profiler cadence)
|
||||
debugCmdC chan string // one-shot debug commands from the cmd-file poller; nil = disabled
|
||||
|
||||
// Per-file write protocol (see requestSave). Workers are a shared pool and
|
||||
// the on-disk staging file is per-file, so two concurrent writes for the
|
||||
// same file would interleave on the temp file; the protocol keeps at most
|
||||
// one write in flight per file and re-issues deferred saves from the write
|
||||
// result, so the rename that lands last always carries the newest content.
|
||||
// All three maps are touched only on the owner goroutine.
|
||||
writeInFlight map[string]int // filename -> fileVersion carried by the in-flight write
|
||||
savePending map[string]bool // filename -> save requested while a write was in flight
|
||||
retryScheduled map[string]time.Time // filename -> armed-but-unfired backoff retry
|
||||
workerPool *pool.WorkerPool
|
||||
mockFS pool.FileSystem
|
||||
done chan struct{}
|
||||
exitWg sync.WaitGroup
|
||||
saveTimer *time.Timer // auto-save debounce timer; non-nil while pending
|
||||
lastEmit time.Time // time of the last frame emission (profiler cadence)
|
||||
debugCmdC chan string // one-shot debug commands from the cmd-file poller; nil = disabled
|
||||
}
|
||||
|
||||
// NewLogic creates a new Logic instance, accepting an optional mockFS.
|
||||
|
|
@ -121,6 +131,9 @@ func NewLogic(mfs pool.FileSystem, path string, openfunc func(string)) *Logic {
|
|||
retryChan: make(chan string, 1), // Buffered channel
|
||||
autosaveChan: make(chan struct{}),
|
||||
inspectChan: make(chan *inspectReq),
|
||||
writeInFlight: make(map[string]int),
|
||||
savePending: make(map[string]bool),
|
||||
retryScheduled: make(map[string]time.Time),
|
||||
workerPool: wp,
|
||||
mockFS: mockFS,
|
||||
done: make(chan struct{}),
|
||||
|
|
@ -192,6 +205,11 @@ func (l *Logic) Run() {
|
|||
for {
|
||||
select {
|
||||
case <-l.done:
|
||||
// Settle in-flight file writes before exiting: the post-exit
|
||||
// synchronous FlushAll and workerPool.Stop must not race a
|
||||
// straggling worker write, whose rename could otherwise land after
|
||||
// the final flush and promote a stale snapshot.
|
||||
l.drainWrites()
|
||||
return
|
||||
case update := <-l.configChan:
|
||||
update.apply(l.state)
|
||||
|
|
@ -232,30 +250,13 @@ func (l *Logic) Run() {
|
|||
l.workerPool.Dispatch(pool.NewStatFileTask(path, l.mockFS))
|
||||
case filename := <-l.retryChan:
|
||||
log.Printf("Logic: Retrying save for %s", filename)
|
||||
if filename == l.state.Editor.Filename {
|
||||
// Reconstruct full content from chunked buffer for saving
|
||||
content, ok := l.fullContentBytes()
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
l.workerPool.DispatchNonBlocking(
|
||||
pool.NewWriteFileTask(filename, content, l.mockFS),
|
||||
)
|
||||
}
|
||||
delete(l.retryScheduled, filename)
|
||||
l.requestSave(filename)
|
||||
case <-l.autosaveChan:
|
||||
// Auto-save debounce tick. The timer goroutine only sent a token;
|
||||
// the owner reconstructs content and dispatches the write.
|
||||
// the owner snapshots content and dispatches the write.
|
||||
l.saveTimer = nil
|
||||
if l.state.Editor.Filename == "" {
|
||||
break
|
||||
}
|
||||
content, ok := l.fullContentBytes()
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
l.workerPool.DispatchNonBlocking(
|
||||
pool.NewWriteFileTask(l.state.Editor.Filename, content, l.mockFS),
|
||||
)
|
||||
l.requestSave(l.state.Editor.Filename)
|
||||
case p := <-l.state.pasteChan:
|
||||
// Clipboard content arrived from the main goroutine: insert it
|
||||
// (replacing any live selection, per the selection-aware edit rule).
|
||||
|
|
@ -304,7 +305,14 @@ func (l *Logic) emitFrame() {
|
|||
}
|
||||
PerfRecord(rec)
|
||||
}
|
||||
l.frameChan <- l.frameOf(elems)
|
||||
select {
|
||||
case l.frameChan <- l.frameOf(elems):
|
||||
default:
|
||||
// Main has not consumed the previous frame yet; drop this one. Frames
|
||||
// are snapshots and the next emission carries the latest state. Keeps
|
||||
// the owner from ever blocking on a slow/gone main (e.g. during the
|
||||
// post-done write drain).
|
||||
}
|
||||
}
|
||||
|
||||
// EnableDebugCmdPoll starts a background poller (debug-only) that watches
|
||||
|
|
@ -500,26 +508,50 @@ func (l *Logic) handleWorkerResult(res pool.Result) {
|
|||
}
|
||||
}
|
||||
} else if res.TaskType == pool.TypeWriteFile {
|
||||
f := res.FilePath
|
||||
wrote, tracked := l.writeInFlight[f]
|
||||
delete(l.writeInFlight, f)
|
||||
pending := l.savePending[f]
|
||||
delete(l.savePending, f)
|
||||
if res.Success {
|
||||
l.state.Editor.lastWriteVersion[res.FilePath] = l.state.Editor.fileVersion[res.FilePath]
|
||||
l.state.Editor.SetWriteFailed(res.FilePath, false)
|
||||
if tracked {
|
||||
// Record the version whose content was actually written (the
|
||||
// snapshot version), not the version now: edits made while the
|
||||
// write was in flight sit ahead of it and trigger the re-issue
|
||||
// below.
|
||||
l.state.Editor.lastWriteVersion[f] = wrote
|
||||
}
|
||||
l.state.Editor.SetWriteFailed(f, false)
|
||||
// Latest-state-wins re-issue: a save was deferred while this write
|
||||
// was in flight, or an edit arrived during it — write the newer
|
||||
// content now (only the active file is reconstructable).
|
||||
if f == l.state.Editor.Filename && (pending || l.state.Editor.fileVersion[f] > l.state.Editor.lastWriteVersion[f]) {
|
||||
l.requestSave(f)
|
||||
}
|
||||
} else {
|
||||
log.Printf("Auto-save failed for %s: %v", res.FilePath, res.Error)
|
||||
l.state.Editor.SetWriteFailed(res.FilePath, true)
|
||||
log.Printf("Auto-save failed for %s: %v", f, res.Error)
|
||||
l.state.Editor.SetWriteFailed(f, true)
|
||||
|
||||
// Trigger retry logic: exponential backoff
|
||||
attempts := l.state.Editor.IncrementRetryAttempts(res.FilePath)
|
||||
attempts := l.state.Editor.IncrementRetryAttempts(f)
|
||||
// Simple backoff: 1s, 2s, 4s, 8s... max 30s
|
||||
delay := time.Duration(1<<(attempts-1)) * time.Second
|
||||
if delay > 30*time.Second {
|
||||
delay = 30 * time.Second
|
||||
}
|
||||
|
||||
filename := res.FilePath
|
||||
l.retryScheduled[f] = time.Now().Add(delay)
|
||||
filename := f
|
||||
log.Printf("Scheduling retry for %s in %v (attempt %d)", filename, delay, attempts)
|
||||
time.AfterFunc(delay, func() {
|
||||
log.Printf("Firing retry for %s", filename)
|
||||
l.retryChan <- filename
|
||||
select {
|
||||
case l.retryChan <- filename:
|
||||
default:
|
||||
// A token for the same file is already queued; it will
|
||||
// re-snapshot the latest content when it fires. Dropping
|
||||
// keeps the timer goroutine from ever blocking.
|
||||
log.Printf("Retry token for %s dropped: one already queued", filename)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -581,6 +613,14 @@ func (l *Logic) WaitForExit() {
|
|||
func (l *Logic) FlushAll() {
|
||||
for filename := range l.state.Editor.fileVersion {
|
||||
if l.state.Editor.IsDirty() && l.state.Editor.Filename == filename {
|
||||
if _, inflight := l.writeInFlight[filename]; inflight {
|
||||
// A worker is already writing this file and we cannot block the
|
||||
// owner (we are it): defer. The write's result handler re-issues
|
||||
// a fresh write while the file is still dirty, so the disk ends
|
||||
// up with the latest content.
|
||||
l.savePending[filename] = true
|
||||
continue
|
||||
}
|
||||
content, ok := l.fullContentBytes()
|
||||
if !ok {
|
||||
continue
|
||||
|
|
@ -592,6 +632,84 @@ func (l *Logic) FlushAll() {
|
|||
}
|
||||
}
|
||||
|
||||
// requestSave is the single entry point for "persist the active file now"
|
||||
// (autosave tick, retry token, deferred re-issue). It implements the
|
||||
// per-file write protocol:
|
||||
//
|
||||
// - at most one write per file is in flight at any time. The worker pool is
|
||||
// shared and the on-disk staging file is per-file, so two concurrent
|
||||
// writes for the same file would interleave on the staging file and could
|
||||
// rename a byte-mixture into place;
|
||||
// - each write carries the file version at the moment its content was
|
||||
// snapshotted (writeInFlight[f]); on success exactly that version is
|
||||
// recorded as written, so any edit that arrived while the write was in
|
||||
// flight leaves the file dirty and the result handler re-issues a fresh
|
||||
// write with the newer content;
|
||||
// - a save requested while a write is in flight is deferred (savePending)
|
||||
// and re-issued by the result handler, so "last rename wins" always
|
||||
// coincides with "newest snapshot wins".
|
||||
//
|
||||
// Must be called on the owner goroutine.
|
||||
func (l *Logic) requestSave(filename string) {
|
||||
if filename == "" || filename != l.state.Editor.Filename {
|
||||
// Only the active file's content lives in memory (its ChunkedBuffer);
|
||||
// a save for any previously opened file cannot be reconstructed.
|
||||
return
|
||||
}
|
||||
if _, inflight := l.writeInFlight[filename]; inflight {
|
||||
l.savePending[filename] = true
|
||||
return
|
||||
}
|
||||
content, ok := l.fullContentBytes()
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
l.writeInFlight[filename] = l.state.Editor.fileVersion[filename]
|
||||
l.workerPool.DispatchNonBlocking(pool.NewWriteFileTask(filename, content, l.mockFS))
|
||||
}
|
||||
|
||||
const (
|
||||
// writeDrainTimeout bounds the post-done wait for in-flight writes so a
|
||||
// pathologically failing write can never hang shutdown.
|
||||
writeDrainTimeout = 5 * time.Second
|
||||
// writeDrainTick is the idle poll interval of the drain loop.
|
||||
writeDrainTick = 20 * time.Millisecond
|
||||
)
|
||||
|
||||
// drainWrites waits for in-flight file writes (and armed retries) to settle
|
||||
// after the done signal, so that Shutdown's post-exit synchronous FlushAll
|
||||
// and workerPool.Stop cannot race a straggling worker write. The write
|
||||
// protocol may re-issue follow-up writes while draining; the deadline bounds
|
||||
// the wait. Must be called on the owner goroutine.
|
||||
func (l *Logic) drainWrites() {
|
||||
if len(l.writeInFlight) == 0 && len(l.retryScheduled) == 0 {
|
||||
return
|
||||
}
|
||||
log.Printf("Logic: draining %d in-flight write(s) before exit", len(l.writeInFlight))
|
||||
deadline := time.Now().Add(writeDrainTimeout)
|
||||
for {
|
||||
if len(l.writeInFlight) == 0 && len(l.retryScheduled) == 0 {
|
||||
return
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
log.Printf("Logic: write drain timed out with %d in flight; exiting (per-write unique temp files keep every rename a complete snapshot)", len(l.writeInFlight))
|
||||
return
|
||||
}
|
||||
select {
|
||||
case res := <-l.workerPool.ResultChan():
|
||||
l.handleWorkerResult(res)
|
||||
case filename := <-l.retryChan:
|
||||
delete(l.retryScheduled, filename)
|
||||
l.requestSave(filename)
|
||||
case <-l.autosaveChan:
|
||||
l.saveTimer = nil
|
||||
l.requestSave(l.state.Editor.Filename)
|
||||
case <-time.After(writeDrainTick):
|
||||
// No events: re-check settle and deadline.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown gracefully shuts down the logic goroutine and worker pool.
|
||||
// The logic goroutine is stopped FIRST so that FlushAll can touch state
|
||||
// single-threaded; the worker pool is stopped last so in-flight results
|
||||
|
|
|
|||
|
|
@ -4,6 +4,10 @@ import (
|
|||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
||||
"pad/internal/io/pool/types"
|
||||
)
|
||||
|
||||
|
|
@ -67,26 +71,70 @@ func (fs *RealFileSystem) ReadFileAt(path string, offset, size int) ([]byte, err
|
|||
}
|
||||
|
||||
func (fs *RealFileSystem) WriteFile(path string, content []byte) error {
|
||||
// Simple write for backward compatibility if needed,
|
||||
// Simple write for backward compatibility if needed,
|
||||
// but defer to Atomic implementation.
|
||||
return fs.WriteFileAtomic(path, content)
|
||||
}
|
||||
|
||||
// tmpSeq numbers the per-call temp files so two writes never share a staging
|
||||
// path. Combined with the owner-side per-file write serialization this makes
|
||||
// same-file write interleaving structurally impossible.
|
||||
var tmpSeq atomic.Uint64
|
||||
|
||||
func (fs *RealFileSystem) WriteFileAtomic(path string, content []byte) error {
|
||||
// Atomic write implementation
|
||||
fullPath := filepath.Join(fs.WorkingDir, path)
|
||||
// Temp file in the same directory as the target to ensure same filesystem rename
|
||||
tmpPath := filepath.Join(filepath.Dir(fullPath), "."+filepath.Base(path)+".tmp")
|
||||
|
||||
// Temp file in the same directory as the target to ensure same filesystem
|
||||
// rename. Unique per call (".<name>.tmp.<pid>.<seq>"): even a concurrent
|
||||
// write or a crashed process's leftover cannot interleave with this one
|
||||
// on the staging file, and rename promotes only this call's complete
|
||||
// content.
|
||||
base := filepath.Base(fullPath)
|
||||
tmpPath := filepath.Join(filepath.Dir(fullPath),
|
||||
"."+base+".tmp."+strconv.Itoa(os.Getpid())+"."+strconv.FormatUint(tmpSeq.Add(1), 10))
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(tmpPath), 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
if err := os.WriteFile(tmpPath, content, 0644); err != nil {
|
||||
os.Remove(tmpPath)
|
||||
return err
|
||||
}
|
||||
|
||||
return os.Rename(tmpPath, fullPath)
|
||||
|
||||
fs.removeStaleTemps(filepath.Dir(fullPath), base, tmpPath)
|
||||
|
||||
if err := os.Rename(tmpPath, fullPath); err != nil {
|
||||
os.Remove(tmpPath)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// removeStaleTemps best-effort removes leftover staging files for the same
|
||||
// file: any ".<base>.tmp.<pid>.<seq>" that is not the caller's own temp, plus
|
||||
// the legacy deterministic name ".<base>.tmp" from older app versions. A
|
||||
// live write never owns either: same-file writes are serialized by the logic
|
||||
// owner, and a live temp always carries the caller's own pid/seq (excluded
|
||||
// via keepTmp). Errors are ignored.
|
||||
func (fs *RealFileSystem) removeStaleTemps(dir, base, keepTmp string) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
keep := filepath.Base(keepTmp)
|
||||
prefix := "." + base + ".tmp."
|
||||
legacy := "." + base + ".tmp"
|
||||
for _, e := range entries {
|
||||
name := e.Name()
|
||||
if name == keep {
|
||||
continue // this call's own live temp file
|
||||
}
|
||||
if name == legacy || strings.HasPrefix(name, prefix) {
|
||||
os.Remove(filepath.Join(dir, name))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (fs *RealFileSystem) DeleteFile(path string) error {
|
||||
|
|
@ -102,6 +150,6 @@ type realDirEntry struct {
|
|||
entry os.DirEntry
|
||||
}
|
||||
|
||||
func (e *realDirEntry) Name() string { return e.entry.Name() }
|
||||
func (e *realDirEntry) IsDir() bool { return e.entry.IsDir() }
|
||||
func (e *realDirEntry) Name() string { return e.entry.Name() }
|
||||
func (e *realDirEntry) IsDir() bool { return e.entry.IsDir() }
|
||||
func (e *realDirEntry) Info() (os.FileInfo, error) { return e.entry.Info() }
|
||||
|
|
|
|||
|
|
@ -170,7 +170,16 @@ func TestWriteFileAtomic_FailedWriteLeavesOriginalIntact(t *testing.T) {
|
|||
func TestWriteFileAtomic_StaleTempFileIsConsumed(t *testing.T) {
|
||||
d := t.TempDir()
|
||||
fs := NewRealFileSystem(d)
|
||||
if err := os.WriteFile(filepath.Join(d, ".f.txt.tmp"), []byte("stale garbage from a crashed write"), 0o644); err != nil {
|
||||
// Stale temps in both the legacy deterministic pattern (older app
|
||||
// versions) and the current unique pattern (another/crashed process).
|
||||
stale := []string{".f.txt.tmp", ".f.txt.tmp.99999.7"}
|
||||
for _, name := range stale {
|
||||
if err := os.WriteFile(filepath.Join(d, name), []byte("stale garbage from a crashed write"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// A different file's temp must be untouched.
|
||||
if err := os.WriteFile(filepath.Join(d, ".other.txt.tmp.99999.8"), []byte("not ours"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fresh := []byte("fresh content")
|
||||
|
|
@ -188,10 +197,15 @@ func TestWriteFileAtomic_StaleTempFileIsConsumed(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, e := range entries {
|
||||
if e.Name() == ".f.txt.tmp" {
|
||||
t.Fatal("stale temp file was not consumed by the successful write")
|
||||
}
|
||||
seen[e.Name()] = true
|
||||
}
|
||||
if seen[".f.txt.tmp"] || seen[".f.txt.tmp.99999.7"] {
|
||||
t.Fatalf("stale temp file(s) were not consumed by the successful write: %v", seen)
|
||||
}
|
||||
if !seen[".other.txt.tmp.99999.8"] {
|
||||
t.Fatalf("another file's temp was removed: %v", seen)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -289,7 +289,9 @@ func TestRealFile_Fuzz_EditSaveReload(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
for _, e := range entries {
|
||||
if strings.HasSuffix(e.Name(), ".tmp") {
|
||||
// Matches both the current ".<name>.tmp.<pid>.<seq>" pattern and the
|
||||
// legacy ".<name>.tmp" one.
|
||||
if strings.Contains(e.Name(), ".tmp") {
|
||||
t.Fatalf("stray temp file left behind: %s", e.Name())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
204
internal/test/e2e/write_serialization_test.go
Normal file
204
internal/test/e2e/write_serialization_test.go
Normal file
|
|
@ -0,0 +1,204 @@
|
|||
package e2e_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"pad/internal/editor"
|
||||
"pad/internal/io/pool"
|
||||
"pad/internal/io/pool/real"
|
||||
"pad/internal/io/pool/types"
|
||||
"pad/internal/test/e2e"
|
||||
)
|
||||
|
||||
// slowCountingFS wraps a pool.FileSystem, delaying each save by delay and
|
||||
// recording the peak number of concurrent saves. A peak > 1 is exactly the
|
||||
// same-file write concurrency that the write protocol must eliminate (two
|
||||
// concurrent writes would otherwise race on the per-file staging path and,
|
||||
// pre-unique-temp, could interleave byte-mixtures onto disk).
|
||||
//
|
||||
// NOTE: the pool's WriteFileTask calls FS.WriteFile (the real FS implements
|
||||
// that as an atomic write via delegation to WriteFileAtomic), while
|
||||
// FlushAll calls WriteFileAtomic directly — so both entry points are
|
||||
// intercepted here, mirroring the real FS's delegation.
|
||||
type slowCountingFS struct {
|
||||
inner pool.FileSystem
|
||||
delay time.Duration
|
||||
|
||||
mu sync.Mutex
|
||||
cur int
|
||||
max int
|
||||
}
|
||||
|
||||
func (f *slowCountingFS) begin() {
|
||||
f.mu.Lock()
|
||||
f.cur++
|
||||
if f.cur > f.max {
|
||||
f.max = f.cur
|
||||
}
|
||||
f.mu.Unlock()
|
||||
time.Sleep(f.delay)
|
||||
}
|
||||
|
||||
func (f *slowCountingFS) end() {
|
||||
f.mu.Lock()
|
||||
f.cur--
|
||||
f.mu.Unlock()
|
||||
}
|
||||
|
||||
func (f *slowCountingFS) current() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.cur
|
||||
}
|
||||
|
||||
func (f *slowCountingFS) peak() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.max
|
||||
}
|
||||
|
||||
func (f *slowCountingFS) ReadDir(path string) ([]types.DirEntry, error) {
|
||||
return f.inner.ReadDir(path)
|
||||
}
|
||||
func (f *slowCountingFS) DirExists(path string) bool {
|
||||
return f.inner.DirExists(path)
|
||||
}
|
||||
func (f *slowCountingFS) FileExists(path string) bool {
|
||||
return f.inner.FileExists(path)
|
||||
}
|
||||
func (f *slowCountingFS) ReadFile(path string) ([]byte, error) {
|
||||
return f.inner.ReadFile(path)
|
||||
}
|
||||
func (f *slowCountingFS) ReadFileAt(path string, offset, size int) ([]byte, error) {
|
||||
return f.inner.ReadFileAt(path, offset, size)
|
||||
}
|
||||
func (f *slowCountingFS) WriteFile(path string, content []byte) error {
|
||||
f.begin()
|
||||
err := f.inner.WriteFile(path, content)
|
||||
f.end()
|
||||
return err
|
||||
}
|
||||
func (f *slowCountingFS) WriteFileAtomic(path string, content []byte) error {
|
||||
f.begin()
|
||||
err := f.inner.WriteFileAtomic(path, content)
|
||||
f.end()
|
||||
return err
|
||||
}
|
||||
func (f *slowCountingFS) DeleteFile(path string) error {
|
||||
return f.inner.DeleteFile(path)
|
||||
}
|
||||
func (f *slowCountingFS) CreateDir(path string) error {
|
||||
return f.inner.CreateDir(path)
|
||||
}
|
||||
|
||||
// waitForDisk polls the on-disk file until it equals want (or the timeout).
|
||||
// The final model is unique, so matching it proves the whole save chain
|
||||
// (deferred re-issues included) has landed.
|
||||
func waitForDisk(t *testing.T, path, want string, timeout time.Duration) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
b, err := os.ReadFile(path)
|
||||
if err == nil && string(b) == want {
|
||||
return
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("disk did not reach %q within %v (last read: %q)", want, timeout, readOrErr(path))
|
||||
}
|
||||
|
||||
func readOrErr(path string) string {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err.Error()
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func newSerializationHarness(t *testing.T, name string, seed string, delay time.Duration) (*slowCountingFS, *e2e.Harness, string) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
diskPath := filepath.Join(dir, name)
|
||||
if err := os.WriteFile(diskPath, []byte(seed), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fs := &slowCountingFS{inner: real.NewRealFileSystem(dir), delay: delay}
|
||||
h := e2e.NewHarness(e2e.WithFileSystem(fs, "/"))
|
||||
h.Run()
|
||||
t.Cleanup(h.Cleanup)
|
||||
loadRealFile(t, h, name)
|
||||
return fs, h, diskPath
|
||||
}
|
||||
|
||||
// TestWriteSerialization_ConcurrentSavesSerialize drives two autosaves whose
|
||||
// 2 s saves would overlap (edit 2 lands 0.5 s after save 1 starts) and
|
||||
// proves the protocol never runs two same-file saves concurrently: save 2's
|
||||
// token is deferred while save 1 is in flight, and the result handler
|
||||
// re-issues it with the newer content. On the pre-fix code the two saves ran
|
||||
// concurrently on the same deterministic temp file (peak 2, possible byte
|
||||
// mixture on disk).
|
||||
func TestWriteSerialization_ConcurrentSavesSerialize(t *testing.T) {
|
||||
fs, h, diskPath := newSerializationHarness(t, "ser.txt", "original content\n", 2*time.Second)
|
||||
|
||||
// Edit 1: autosave debounce fires ~1 s later -> save W1 (in flight
|
||||
// ~1.0 s .. ~3.0 s).
|
||||
if err := h.WithState(func(st *editor.State) {
|
||||
st.Editor.CursorPosition = 0
|
||||
editor.HandleInsert("A")
|
||||
}); err != nil {
|
||||
t.Fatalf("edit 1: %v", err)
|
||||
}
|
||||
time.Sleep(1500 * time.Millisecond) // W1 is in flight now
|
||||
|
||||
// Edit 2 while W1 is in flight: its autosave token fires ~2.5 s, inside
|
||||
// W1's window -> must be deferred, not dispatched concurrently.
|
||||
if err := h.WithState(func(st *editor.State) {
|
||||
st.Editor.CursorPosition = 1
|
||||
editor.HandleInsert("B")
|
||||
}); err != nil {
|
||||
t.Fatalf("edit 2: %v", err)
|
||||
}
|
||||
|
||||
const model = "ABoriginal content\n"
|
||||
waitForDisk(t, diskPath, model, 20*time.Second)
|
||||
|
||||
if peak := fs.peak(); peak > 1 {
|
||||
t.Fatalf("peak concurrent same-file saves = %d, want <= 1", peak)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWriteSerialization_FlushDuringInFlight drives the GoToBrowser/Shutdown
|
||||
// path (FlushAll) while a worker save is in flight: the flush must defer
|
||||
// (no concurrent save), and the in-flight save's completion must re-issue the
|
||||
// latest content so "last rename wins" coincides with "newest snapshot wins".
|
||||
func TestWriteSerialization_FlushDuringInFlight(t *testing.T) {
|
||||
fs, h, diskPath := newSerializationHarness(t, "flushser.txt", "base\n", 2*time.Second)
|
||||
|
||||
// Edit -> autosave token ~1 s -> save W1 (in flight ~1.0 s .. ~3.0 s).
|
||||
if err := h.WithState(func(st *editor.State) {
|
||||
st.Editor.CursorPosition = 0
|
||||
editor.HandleInsert("X")
|
||||
}); err != nil {
|
||||
t.Fatalf("edit: %v", err)
|
||||
}
|
||||
time.Sleep(1500 * time.Millisecond) // W1 is in flight now
|
||||
|
||||
// Flush while W1 is in flight.
|
||||
if err := h.Flush(); err != nil {
|
||||
t.Fatalf("flush: %v", err)
|
||||
}
|
||||
if cur := fs.current(); cur != 1 {
|
||||
t.Fatalf("Flush during in-flight save: concurrent saves = %d, want 1 (the in-flight one only)", cur)
|
||||
}
|
||||
|
||||
const model = "Xbase\n"
|
||||
waitForDisk(t, diskPath, model, 20*time.Second)
|
||||
|
||||
if peak := fs.peak(); peak > 1 {
|
||||
t.Fatalf("peak concurrent same-file saves = %d, want <= 1", peak)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user