From 9db81e4739ec52cc41e7a686a76f96e2e698bc8e Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 1 Oct 2026 17:14:23 +0300 Subject: [PATCH 1/6] Fix concurrent same-content transfers, disk-full handling and small-frame writes Found by running a node in a 1 GB container under load. - Several transfers of the same content at once failed: the .partial is named after the content hash, so they shared one file and all but the first failed at the final rename. The first transfer keeps the resumable .partial; a concurrent one gets a private file, removed if it does not finish. - A file that cannot fit on the receiver's disk is refused at INIT (16 MiB kept in reserve) instead of running until the disk is full. If a write still hits ENOSPC the .partial is deleted rather than left holding the disk full. - A refusal at INIT is reported as a refusal. The sender took it for a receiver without streamed-transfer support, and the caller fell back to pushing the whole file in one frame. - WriteFrame sends a frame of up to 64 KiB as one write, so the payload is not held behind the header by Nagle and the peer's delayed ACK. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 18 +++ dataexchange.go | 16 +++ filestream.go | 171 +++++++++++++++++++++-- filestream_disk_other.go | 9 ++ filestream_disk_unix.go | 17 +++ zz_filestream_concurrent_test.go | 227 +++++++++++++++++++++++++++++++ zz_service_errors_test.go | 3 +- 7 files changed, 445 insertions(+), 16 deletions(-) create mode 100644 filestream_disk_other.go create mode 100644 filestream_disk_unix.go create mode 100644 zz_filestream_concurrent_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 06841b3..d6f8414 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Several transfers of the same content at once no longer fail. The receiver + names a transfer's `.partial` after the content hash so a retry can resume + it; two peers sending the same file (or one sender on two connections) + therefore wrote the same file, and all but the first failed at the final + rename. The first transfer keeps the resumable `.partial`; a concurrent one + gets a private file that is removed if it does not finish. +- A file that cannot fit on the receiver's disk is refused before any byte is + sent. The byte quota is a fixed number and can be larger than the disk, so a + transfer ran until the disk was full, failed, and left its bytes in + `.partial` — where they kept the disk full and blocked incoming messages. + The receiver now checks free space at the start (keeping 16 MiB in reserve), + and if a write still hits a full disk it deletes that `.partial`. +- A sender whose transfer is refused at the start (quota, disk full, too many + transfers) reports the refusal. It was mistaken for a receiver too old to + support streamed transfers, so the caller retried with the single-frame + path and pushed the whole file at a peer that had just declined it. +- `WriteFrame` sends a frame of up to 64 KiB as one write. Header and payload + as two writes made the payload wait on Nagle and the peer's delayed ACK. - The inbox byte cap no longer deletes the whole inbox. With the default config (`InboxMaxBytes == 0`, meaning 256 MiB) the evictor compared the inbox size against the raw value 0, so the first time the inbox passed diff --git a/dataexchange.go b/dataexchange.go index 8bffc8b..6f2def2 100644 --- a/dataexchange.go +++ b/dataexchange.go @@ -195,6 +195,17 @@ func WriteFrame(w io.Writer, f *Frame) error { var hdr [8]byte binary.BigEndian.PutUint32(hdr[0:4], ftype) binary.BigEndian.PutUint32(hdr[4:8], uint32(len(payload))) + // A small frame goes out as one write. Written as header-then-payload, + // the payload is a second small write on a stream that still has the + // header unacknowledged: Nagle holds it until the peer's delayed ACK + // fires, ~40ms per frame on daemons that delay ACKs for short segments. + if len(payload) <= singleWriteMax { + buf := make([]byte, len(hdr)+len(payload)) + copy(buf, hdr[:]) + copy(buf[len(hdr):], payload) + _, err = w.Write(buf) + return err + } if _, err := w.Write(hdr[:]); err != nil { return err } @@ -202,6 +213,11 @@ func WriteFrame(w io.Writer, f *Frame) error { return err } +// singleWriteMax is the largest payload WriteFrame copies next to its header +// to send both in one write. Above it the copy costs more than it saves: the +// payload spans many segments and is not held back by Nagle. +const singleWriteMax = 64 * 1024 + // ReadFrame reads a frame from a reader. func ReadFrame(r io.Reader) (*Frame, error) { var hdr [8]byte diff --git a/filestream.go b/filestream.go index e200d89..a8dfa1f 100644 --- a/filestream.go +++ b/filestream.go @@ -28,6 +28,7 @@ package dataexchange import ( "bytes" + "crypto/rand" "crypto/sha256" "encoding/binary" "encoding/hex" @@ -38,6 +39,7 @@ import ( "path/filepath" "sync" "sync/atomic" + "syscall" "time" ) @@ -248,6 +250,19 @@ func streamSendWithInit(conn frameRW, name string, r io.ReadSeeker, size int64, return nil, ErrStreamUnsupported } kind, gotID, body, ok := decodeStreamFrame(initAck) + if ok && kind == streamKindComplete && gotID == id { + // The receiver speaks the protocol and said no (quota, disk full, + // too many transfers). That is a refusal to report, not an old + // receiver to retry against with the single-frame path — which + // would push the whole file at a peer that just declined it. + _, msg := decodeComplete(body) + return &StreamResult{ + TotalBytes: size, + Sha256: hex.EncodeToString(fullHash[:]), + OK: false, + Message: msg, + }, nil + } if !ok || kind != streamKindInitAck || gotID != id { // A legacy receiver answers with a plain "ACK UNKNOWN(7)" TEXT // frame, or nothing useful — treat as unsupported. @@ -433,8 +448,12 @@ type StreamReceiver struct { } type recvTransfer struct { - file *os.File - partial string + file *os.File + partial string + // private marks a .partial that only this transfer can use: another + // live transfer already held the content-addressed one, so this one got + // its own. Nothing can resume it, so it is removed if it does not finish. + private bool name string size uint64 hash [32]byte @@ -448,6 +467,53 @@ type recvTransfer struct { quotaBudget int64 } +// activePartials is the set of .partial paths a live transfer is writing. +// +// A .partial is named after the content hash so a retry resumes it. That also +// means two transfers of the same content at once — two peers sending the +// same file, or one sender on two connections — would open the same path, +// interleave their writes, and the second to finish would fail its rename +// because the first already moved the file. The first transfer to arrive +// claims the path; a concurrent one writes to a private .partial instead. +var activePartials = struct { + sync.Mutex + paths map[string]struct{} +}{paths: make(map[string]struct{})} + +func claimPartial(path string) bool { + activePartials.Lock() + defer activePartials.Unlock() + if _, held := activePartials.paths[path]; held { + return false + } + activePartials.paths[path] = struct{}{} + return true +} + +func releasePartial(path string) { + activePartials.Lock() + delete(activePartials.paths, path) + activePartials.Unlock() +} + +// release gives up t's claim on its .partial and, for a private one, removes +// the file (a no-op once a finished transfer has renamed it into place). +func (t *recvTransfer) release() { + releasePartial(t.partial) + if t.private { + _ = os.Remove(t.partial) + } +} + +// diskReserveBytes is left free on the receiving filesystem: a transfer that +// would take the disk below it is refused at INIT, so a large file cannot +// leave the node unable to store messages or its own state. +const diskReserveBytes = 16 << 20 + +// freeDiskBytes reports the space available on the filesystem holding dir; +// ok is false where that cannot be determined. A variable so tests can set it. +var freeDiskBytes = platformFreeDiskBytes + // NewStreamReceiver builds a receiver writing into receivedDir. nameSuffix // maps a base filename to a final unique name (nil ⇒ a timestamped default). // onSaved (nil ok) fires after a verified file is renamed into place. No disk @@ -552,6 +618,36 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame return encodeComplete(id, false, "mkdir partial: "+err.Error()) } partial := filepath.Join(partialDir, hex.EncodeToString(id[:])) + private := false + sr.mu.Lock() + prior := sr.transfers[id] + sr.mu.Unlock() + switch { + case prior != nil: + // A second INIT for a transfer this receiver already holds: keep + // its .partial (and the claim on it). + partial, private = prior.partial, prior.private + case !claimPartial(partial): + // Another live transfer is writing this content. Take a private + // .partial rather than share the file with it. + var suffix [4]byte + if _, err := rand.Read(suffix[:]); err != nil { + return encodeComplete(id, false, "partial name: "+err.Error()) + } + partial += "." + hex.EncodeToString(suffix[:]) + private = true + claimPartial(partial) + } + // fail reports a refused INIT, giving back the claim just taken. + fail := func(reason string) *Frame { + if prior == nil { + releasePartial(partial) + if private { + _ = os.Remove(partial) + } + } + return encodeComplete(id, false, reason) + } // Quota gate: reject up front if the declared size cannot fit alongside // what is already on disk (excluding this transfer's own .partial, which @@ -564,16 +660,15 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame budget = 0 } if int64(size) > budget { - return encodeComplete(id, false, - fmt.Sprintf("receiver disk quota exceeded: need %d, %d of %d available", - size, budget, sr.quotaBytes)) + return fail(fmt.Sprintf("receiver disk quota exceeded: need %d, %d of %d available", + size, budget, sr.quotaBytes)) } quotaBudget = budget } file, err := os.OpenFile(partial, os.O_RDWR|os.O_CREATE, 0600) if err != nil { - return encodeComplete(id, false, "open partial: "+err.Error()) + return fail("open partial: " + err.Error()) } // Resume from the contiguous bytes already on disk. Because we only // ever write contiguously, the file size IS the resume offset. Guard @@ -587,6 +682,21 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame resume = 0 } + // Disk gate: the quota is a fixed number and can be larger than the + // disk. Refuse now what cannot fit, instead of failing mid-transfer with + // the disk full and the bytes already written stranded in a .partial. + if free, ok := freeDiskBytes(partialDir); ok { + need := size - resume + if need > free || free-need < diskReserveBytes { + _ = file.Close() + if resume == 0 { + _ = os.Remove(partial) + } + return fail(fmt.Sprintf("receiver disk full: need %d bytes, %d free (keeping %d in reserve)", + need, free, int64(diskReserveBytes))) + } + } + sr.mu.Lock() // Replace any stale in-memory transfer for this id (e.g. a prior conn). if old := sr.transfers[id]; old != nil { @@ -599,11 +709,12 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame // rejection would leak the very FD it exists to protect. sr.mu.Unlock() _ = file.Close() - return encodeComplete(id, false, "too many concurrent transfers") + return fail("too many concurrent transfers") } sr.transfers[id] = &recvTransfer{ file: file, partial: partial, + private: private, name: sanitizeBase(name), size: size, hash: hash, @@ -640,6 +751,7 @@ func (sr *StreamReceiver) handleChunk(id [transferIDLen]byte, body []byte) *Fram case off == t.cursor: if werr := sr.writeAt(t, off, data); werr != nil { sr.mu.Unlock() + sr.abandonIfDiskFull(id, werr) return encodeComplete(id, false, werr.Error()) } // Drain any buffered successors. @@ -652,6 +764,7 @@ func (sr *StreamReceiver) handleChunk(id [transferIDLen]byte, body []byte) *Fram t.pendBytes -= len(next) if werr := sr.writeAt(t, t.cursor, next); werr != nil { sr.mu.Unlock() + sr.abandonIfDiskFull(id, werr) return encodeComplete(id, false, werr.Error()) } } @@ -753,33 +866,61 @@ func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { return encodeComplete(id, true, "") } -// discard closes and forgets a transfer but leaves the .partial on disk. +// discard closes and forgets a transfer but leaves the .partial on disk +// (unless it is a private one, which nothing could resume). func (sr *StreamReceiver) discard(id [transferIDLen]byte) { sr.mu.Lock() - if t := sr.transfers[id]; t != nil && t.file != nil { - _ = t.file.Close() - } + t := sr.transfers[id] delete(sr.transfers, id) sr.mu.Unlock() + if t != nil { + if t.file != nil { + _ = t.file.Close() + } + t.release() + } } func (sr *StreamReceiver) forget(id [transferIDLen]byte) { sr.mu.Lock() + t := sr.transfers[id] delete(sr.transfers, id) sr.mu.Unlock() + if t != nil { + t.release() + } +} + +// abandonIfDiskFull ends a transfer whose write failed for lack of space and +// deletes its .partial. Keeping it would hold the disk full — the receiver +// could then store neither messages nor its own state — for the sake of a +// resume that cannot succeed until space is freed anyway. +func (sr *StreamReceiver) abandonIfDiskFull(id [transferIDLen]byte, werr error) { + if !errors.Is(werr, syscall.ENOSPC) { + return + } + sr.mu.Lock() + t := sr.transfers[id] + sr.mu.Unlock() + sr.discard(id) + if t != nil { + _ = os.Remove(t.partial) + } } // Close releases any open .partial handles (call on connection teardown). -// The .partial files themselves remain for resume. +// The .partial files themselves remain for resume, except private ones. func (sr *StreamReceiver) Close() { sr.mu.Lock() - for _, t := range sr.transfers { + transfers := sr.transfers + sr.transfers = make(map[[transferIDLen]byte]*recvTransfer) + sr.mu.Unlock() + for _, t := range transfers { if t.file != nil { _ = t.file.Close() } + t.release() } - sr.transfers = make(map[[transferIDLen]byte]*recvTransfer) - sr.mu.Unlock() } // dirSizeExcluding sums the on-disk size of every regular file under dir, diff --git a/filestream_disk_other.go b/filestream_disk_other.go new file mode 100644 index 0000000..f9a678d --- /dev/null +++ b/filestream_disk_other.go @@ -0,0 +1,9 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !unix + +package dataexchange + +// platformFreeDiskBytes is not implemented on this platform; the receiver relies on +// its byte quota alone. +func platformFreeDiskBytes(string) (uint64, bool) { return 0, false } diff --git a/filestream_disk_unix.go b/filestream_disk_unix.go new file mode 100644 index 0000000..ffbf436 --- /dev/null +++ b/filestream_disk_unix.go @@ -0,0 +1,17 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build unix + +package dataexchange + +import "syscall" + +// platformFreeDiskBytes reports the space available to this process on the +// filesystem holding dir. +func platformFreeDiskBytes(dir string) (uint64, bool) { + var st syscall.Statfs_t + if err := syscall.Statfs(dir, &st); err != nil { + return 0, false + } + return uint64(st.Bavail) * uint64(st.Bsize), true // #nosec G115 -- block size is positive +} diff --git a/zz_filestream_concurrent_test.go b/zz_filestream_concurrent_test.go new file mode 100644 index 0000000..2b21396 --- /dev/null +++ b/zz_filestream_concurrent_test.go @@ -0,0 +1,227 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +import ( + "bytes" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" +) + +func partialFiles(t *testing.T, dir string) []string { + t.Helper() + entries, err := os.ReadDir(filepath.Join(dir, ".partial")) + if err != nil && !os.IsNotExist(err) { + t.Fatalf("read .partial: %v", err) + } + var names []string + for _, e := range entries { + names = append(names, e.Name()) + } + return names +} + +// The .partial is named after the content hash, so several transfers of the +// same content at once used to share one file: all but the first failed at +// the final rename. Each now completes and leaves its own copy. +func TestFileStream_SameContentConcurrently(t *testing.T) { + dir := t.TempDir() + data := makePayload(3*StreamChunkSize + 4321) + var seq atomic.Int64 + name := func(base string) string { return fmt.Sprintf("%d-%s", seq.Add(1), base) } + + const senders = 4 + var wg sync.WaitGroup + results := make([]*StreamResult, senders) + errs := make([]error, senders) + for i := 0; i < senders; i++ { + cli, srv := net.Pipe() + go func() { + sr := NewStreamReceiver(dir, name, nil) + defer sr.Close() + for { + f, err := ReadFrame(srv) + if err != nil { + return + } + if resp := sr.HandleFrame(f); resp != nil { + if WriteFrame(srv, resp) != nil { + return + } + } + } + }() + wg.Add(1) + go func(i int) { + defer wg.Done() + defer cli.Close() + results[i], errs[i] = streamSend(cli, "same.bin", bytes.NewReader(data), int64(len(data)), 10*time.Second) + }(i) + } + wg.Wait() + + for i := range results { + if errs[i] != nil { + t.Fatalf("sender %d: %v", i, errs[i]) + } + if !results[i].OK { + t.Errorf("sender %d refused: %s", i, results[i].Message) + } + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + files := 0 + for _, e := range entries { + if e.IsDir() { + continue + } + files++ + got, err := os.ReadFile(filepath.Join(dir, e.Name())) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, data) { + t.Errorf("%s does not match the sent content", e.Name()) + } + } + if files != senders { + t.Errorf("received %d files, want %d", files, senders) + } + // The receivers close when their pipes do; give that a moment. + deadline := time.Now().Add(2 * time.Second) + for len(partialFiles(t, dir)) > 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if left := partialFiles(t, dir); len(left) > 0 { + t.Errorf("partial files left behind: %v", left) + } +} + +// A transfer that cannot fit on the disk is refused before any byte lands, +// rather than failing part-way with the disk full. +func TestFileStream_RefusedWhenDiskCannotHoldIt(t *testing.T) { + dir := t.TempDir() + data := makePayload(2 * StreamChunkSize) + prev := freeDiskBytes + freeDiskBytes = func(string) (uint64, bool) { return uint64(len(data)) + diskReserveBytes - 1, true } + defer func() { freeDiskBytes = prev }() + + cli, srv := net.Pipe() + done := make(chan error, 1) + go runStreamReceiver(srv, dir, done) + res, err := streamSend(cli, "big.bin", bytes.NewReader(data), int64(len(data)), 5*time.Second) + _ = cli.Close() + <-done + + if err == nil && res != nil && res.OK { + t.Fatal("transfer accepted with too little free space") + } + msg := fmt.Sprint(err) + if res != nil { + msg += " " + res.Message + } + if !strings.Contains(msg, "receiver disk full") { + t.Errorf("refusal does not say why: %q", msg) + } + if left := partialFiles(t, dir); len(left) > 0 { + t.Errorf("refused transfer left partial files: %v", left) + } + + // With room to spare the same transfer goes through. + freeDiskBytes = func(string) (uint64, bool) { return uint64(len(data)) + diskReserveBytes, true } + cli, srv = net.Pipe() + go runStreamReceiver(srv, dir, done) + res, err = streamSend(cli, "big.bin", bytes.NewReader(data), int64(len(data)), 5*time.Second) + _ = cli.Close() + <-done + if err != nil || !res.OK { + t.Fatalf("transfer with enough space failed: %v %+v", err, res) + } +} + +// When a write fails because the disk is full, the .partial goes: keeping it +// holds the disk full for a resume that cannot succeed. +func TestFileStream_DiskFullMidTransferRemovesPartial(t *testing.T) { + dir := t.TempDir() + data := makePayload(StreamChunkSize) + id := transferIDOf(data) + sr := NewStreamReceiver(dir, nil, nil) + defer sr.Close() + + cli, srv := net.Pipe() + go func() { + for { + f, err := ReadFrame(srv) + if err != nil { + return + } + if f.Type == TypeFileStream && len(f.Payload) > 0 && f.Payload[0] == streamKindInit { + _ = WriteFrame(srv, sr.HandleFrame(f)) + return + } + } + }() + go func() { _, _ = streamSend(cli, "f.bin", bytes.NewReader(data), int64(len(data)), time.Second) }() + deadline := time.Now().Add(2 * time.Second) + for len(partialFiles(t, dir)) == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if len(partialFiles(t, dir)) != 1 { + t.Fatalf("expected one partial after INIT, have %v", partialFiles(t, dir)) + } + + sr.abandonIfDiskFull(id, fmt.Errorf("write at 0: %w", os.ErrPermission)) + if len(partialFiles(t, dir)) != 1 { + t.Fatal("a non-disk-full error must keep the partial for resume") + } + sr.abandonIfDiskFull(id, fmt.Errorf("write at 0: %w", syscall.ENOSPC)) + if left := partialFiles(t, dir); len(left) != 0 { + t.Errorf("partial kept after disk-full: %v", left) + } + _ = cli.Close() + _ = srv.Close() +} + +type writeCounter struct { + bytes.Buffer + writes int +} + +func (c *writeCounter) Write(p []byte) (int, error) { + c.writes++ + return c.Buffer.Write(p) +} + +// A small frame must reach the stream in one write: header and payload as +// two writes stalls on Nagle and the peer's delayed ACK. +func TestWriteFrame_SmallFrameIsOneWrite(t *testing.T) { + for _, tc := range []struct { + size, writes int + }{{0, 1}, {200, 1}, {singleWriteMax, 1}, {singleWriteMax + 1, 2}} { + var w writeCounter + in := &Frame{Type: TypeBinary, Payload: makePayload(tc.size)} + if err := WriteFrame(&w, in); err != nil { + t.Fatal(err) + } + if w.writes != tc.writes { + t.Errorf("%d-byte payload: %d writes, want %d", tc.size, w.writes, tc.writes) + } + out, err := ReadFrame(&w.Buffer) + if err != nil { + t.Fatalf("%d-byte payload: read back: %v", tc.size, err) + } + if out.Type != in.Type || !bytes.Equal(out.Payload, in.Payload) { + t.Errorf("%d-byte payload did not round-trip", tc.size) + } + } +} diff --git a/zz_service_errors_test.go b/zz_service_errors_test.go index e4d8b9d..2d53f29 100644 --- a/zz_service_errors_test.go +++ b/zz_service_errors_test.go @@ -122,10 +122,11 @@ func TestWriteFrame_HeaderWriteError(t *testing.T) { // TestWriteFrame_PayloadWriteError forces the SECOND w.Write to fail (the // payload), exercising the path where the header lands but the body errors. +// Only a payload above singleWriteMax is written separately from its header. func TestWriteFrame_PayloadWriteError(t *testing.T) { t.Parallel() fw := &failingWriter{failAfter: 1} // first write OK, second fails - err := WriteFrame(fw, &Frame{Type: TypeText, Payload: []byte("payload")}) + err := WriteFrame(fw, &Frame{Type: TypeText, Payload: make([]byte, singleWriteMax+1)}) if err == nil { t.Error("expected payload-write error") } From c6b706d75a66b0636f232d25e99acba940ae86fc Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:05:01 +0300 Subject: [PATCH 2/6] filestream: a retry takes over a stalled transfer's .partial; disk check builds everywhere Review of this branch found two regressions against main. A sender that stalls and retries reconnects while the receiver still holds the stalled connection open (idle timeout, minutes). The retry found the content-addressed .partial claimed and started over at offset 0 in a private file, orphaning the stale .partial, which then counted against the quota and could get the retry refused. A claim whose holder has written nothing for staleClaimAfter (10s) can now be taken over: the retry resumes the .partial where the stalled transfer stopped, and the stalled transfer is refused any further write or rename. Two live transfers of the same content still get separate files. //go:build unix included openbsd, netbsd, solaris and illumos, whose syscall.Statfs_t lacks the fields used; the free-space check is now built for linux, darwin and freebsd and other platforms rely on the byte quota. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 +- filestream.go | 84 +++++++++++++++++++++++----- filestream_disk_other.go | 2 +- filestream_disk_unix.go | 6 +- zz_filestream_concurrent_test.go | 94 ++++++++++++++++++++++++++++++++ 5 files changed, 172 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d6f8414..bb3fdb4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,7 +35,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 it; two peers sending the same file (or one sender on two connections) therefore wrote the same file, and all but the first failed at the final rename. The first transfer keeps the resumable `.partial`; a concurrent one - gets a private file that is removed if it does not finish. + gets a private file that is removed if it does not finish. A sender that + stalls and retries still resumes: once the transfer holding the `.partial` + has written nothing for 10 seconds, the retry takes the file over and the + stalled transfer is refused any further write. - A file that cannot fit on the receiver's disk is refused before any byte is sent. The byte quota is a fixed number and can be larger than the disk, so a transfer ran until the disk was full, failed, and left its bytes in diff --git a/filestream.go b/filestream.go index a8dfa1f..6af376c 100644 --- a/filestream.go +++ b/filestream.go @@ -453,7 +453,11 @@ type recvTransfer struct { // private marks a .partial that only this transfer can use: another // live transfer already held the content-addressed one, so this one got // its own. Nothing can resume it, so it is removed if it does not finish. - private bool + private bool + // claim is this transfer's hold on partial (see activePartials). A retry + // can take a stalled transfer's claim over; the old transfer then stops + // writing. + claim *partialClaim name string size uint64 hash [32]byte @@ -475,36 +479,77 @@ type recvTransfer struct { // interleave their writes, and the second to finish would fail its rename // because the first already moved the file. The first transfer to arrive // claims the path; a concurrent one writes to a private .partial instead. +// +// A claim whose holder has written nothing for staleClaimAfter can be taken +// over. That is a sender retrying after a stall: it reconnects while this +// receiver still holds the stalled connection open (until the idle timeout, +// minutes later). The retry takes the .partial over and resumes where the +// stalled transfer stopped; the stalled one is refused any further write. var activePartials = struct { sync.Mutex - paths map[string]struct{} -}{paths: make(map[string]struct{})} + paths map[string]*partialClaim +}{paths: make(map[string]*partialClaim)} + +// partialClaim is one transfer's hold on a .partial path. +type partialClaim struct { + lastWrite time.Time +} -func claimPartial(path string) bool { +// staleClaimAfter is how long a claim's holder may go without writing before +// another transfer of the same content may take the claim over. A variable so +// tests can shorten it. +var staleClaimAfter = 10 * time.Second + +// claimPartial takes path for a new transfer. It returns nil when a live +// transfer holds it. +func claimPartial(path string) *partialClaim { activePartials.Lock() defer activePartials.Unlock() - if _, held := activePartials.paths[path]; held { + if held, ok := activePartials.paths[path]; ok && time.Since(held.lastWrite) < staleClaimAfter { + return nil + } + c := &partialClaim{lastWrite: time.Now()} + activePartials.paths[path] = c + return c +} + +// stillHolds reports whether c is still the claim on path, and records a +// write by its holder when it is. +func stillHolds(path string, c *partialClaim) bool { + if c == nil { + return true // a transfer that never needed a claim + } + activePartials.Lock() + defer activePartials.Unlock() + if activePartials.paths[path] != c { return false } - activePartials.paths[path] = struct{}{} + c.lastWrite = time.Now() return true } -func releasePartial(path string) { +// releasePartial gives up c's claim on path, if it still holds it. +func releasePartial(path string, c *partialClaim) { activePartials.Lock() - delete(activePartials.paths, path) + if activePartials.paths[path] == c { + delete(activePartials.paths, path) + } activePartials.Unlock() } // release gives up t's claim on its .partial and, for a private one, removes // the file (a no-op once a finished transfer has renamed it into place). func (t *recvTransfer) release() { - releasePartial(t.partial) + releasePartial(t.partial, t.claim) if t.private { _ = os.Remove(t.partial) } } +// errPartialTakenOver is returned to a stalled transfer whose .partial a +// retry of the same transfer has taken over. +var errPartialTakenOver = errors.New("a newer transfer of this file has taken it over") + // diskReserveBytes is left free on the receiving filesystem: a transfer that // would take the disk below it is refused at INIT, so a large file cannot // leave the node unable to store messages or its own state. @@ -619,15 +664,15 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame } partial := filepath.Join(partialDir, hex.EncodeToString(id[:])) private := false + var claim *partialClaim sr.mu.Lock() prior := sr.transfers[id] sr.mu.Unlock() - switch { - case prior != nil: + if prior != nil { // A second INIT for a transfer this receiver already holds: keep // its .partial (and the claim on it). - partial, private = prior.partial, prior.private - case !claimPartial(partial): + partial, private, claim = prior.partial, prior.private, prior.claim + } else if claim = claimPartial(partial); claim == nil { // Another live transfer is writing this content. Take a private // .partial rather than share the file with it. var suffix [4]byte @@ -636,12 +681,12 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame } partial += "." + hex.EncodeToString(suffix[:]) private = true - claimPartial(partial) + claim = claimPartial(partial) } // fail reports a refused INIT, giving back the claim just taken. fail := func(reason string) *Frame { if prior == nil { - releasePartial(partial) + releasePartial(partial, claim) if private { _ = os.Remove(partial) } @@ -715,6 +760,7 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame file: file, partial: partial, private: private, + claim: claim, name: sanitizeBase(name), size: size, hash: hash, @@ -788,6 +834,9 @@ func (sr *StreamReceiver) handleChunk(id [transferIDLen]byte, body []byte) *Fram // the write if it would overrun, so a peer that lies about its declared size // (sends more chunk bytes than INIT promised) still cannot exceed the quota. func (sr *StreamReceiver) writeAt(t *recvTransfer, off uint64, data []byte) error { + if !stillHolds(t.partial, t.claim) { + return errPartialTakenOver + } if t.quotaBudget >= 0 { if int64(len(data)) > t.quotaBudget { return fmt.Errorf("receiver disk quota exceeded at offset %d", off) @@ -808,6 +857,11 @@ func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { if t == nil { return encodeComplete(id, false, "no active transfer") } + if !stillHolds(t.partial, t.claim) { + // A retry of this transfer has taken the file over; it finishes it. + sr.forget(id) + return encodeComplete(id, false, errPartialTakenOver.Error()) + } if err := t.file.Sync(); err != nil { return encodeComplete(id, false, "fsync: "+err.Error()) diff --git a/filestream_disk_other.go b/filestream_disk_other.go index f9a678d..194e945 100644 --- a/filestream_disk_other.go +++ b/filestream_disk_other.go @@ -1,6 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -//go:build !unix +//go:build !linux && !darwin && !freebsd package dataexchange diff --git a/filestream_disk_unix.go b/filestream_disk_unix.go index ffbf436..6545a78 100644 --- a/filestream_disk_unix.go +++ b/filestream_disk_unix.go @@ -1,13 +1,15 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -//go:build unix +//go:build linux || darwin || freebsd package dataexchange import "syscall" // platformFreeDiskBytes reports the space available to this process on the -// filesystem holding dir. +// filesystem holding dir. Limited to the platforms whose syscall.Statfs_t has +// Bavail and Bsize; elsewhere (openbsd, netbsd, solaris, illumos, ...) the +// receiver relies on its byte quota alone. func platformFreeDiskBytes(dir string) (uint64, bool) { var st syscall.Statfs_t if err := syscall.Statfs(dir, &st); err != nil { diff --git a/zz_filestream_concurrent_test.go b/zz_filestream_concurrent_test.go index 2b21396..c80f3b2 100644 --- a/zz_filestream_concurrent_test.go +++ b/zz_filestream_concurrent_test.go @@ -4,6 +4,7 @@ package dataexchange import ( "bytes" + "crypto/sha256" "fmt" "net" "os" @@ -225,3 +226,96 @@ func TestWriteFrame_SmallFrameIsOneWrite(t *testing.T) { } } } + +// A sender that stalls and retries reconnects while this receiver still +// holds the stalled connection open (until its idle timeout, minutes later). +// The retry must resume the .partial where the stalled transfer stopped — +// not start over in a private copy, which also left the stale .partial to +// count against the quota and could get the retry refused outright — and +// the stalled transfer must not write to the file again. +func TestFileStream_RetryTakesOverAStalledTransfer(t *testing.T) { + prev := staleClaimAfter + staleClaimAfter = 50 * time.Millisecond + defer func() { staleClaimAfter = prev }() + + dir := t.TempDir() + data := makePayload(3 * StreamChunkSize) + hash := sha256.Sum256(data) + var id [transferIDLen]byte + copy(id[:], hash[:]) + + send := func(sr *StreamReceiver, f *Frame) (kind byte, body []byte) { + t.Helper() + resp := sr.HandleFrame(f) + if resp == nil { + t.Fatal("no response") + } + kind, _, body, ok := decodeStreamFrame(resp) + if !ok { + t.Fatal("malformed response") + } + return kind, body + } + chunk := func(i int) *Frame { + return encodeChunk(id, uint64(i*StreamChunkSize), data[i*StreamChunkSize:(i+1)*StreamChunkSize]) + } + + stalled := NewStreamReceiver(dir, nil, nil) + defer stalled.Close() + if kind, _ := send(stalled, encodeInit(id, uint64(len(data)), hash, StreamChunkSize, "f.bin")); kind != streamKindInitAck { + t.Fatalf("first INIT answered with kind %#x", kind) + } + if kind, _ := send(stalled, chunk(0)); kind != streamKindAck { + t.Fatalf("first chunk answered with kind %#x", kind) + } + time.Sleep(2 * staleClaimAfter) // the sender stalls, then retries + + retry := NewStreamReceiver(dir, nil, nil) + defer retry.Close() + kind, body := send(retry, encodeInit(id, uint64(len(data)), hash, StreamChunkSize, "f.bin")) + if kind != streamKindInitAck { + t.Fatalf("retry INIT answered with kind %#x", kind) + } + if off, _ := decodeOffset(body); off != StreamChunkSize { + t.Fatalf("retry resumes at %d, want %d: it did not take over the stalled transfer's .partial", off, StreamChunkSize) + } + + // The stalled connection wakes up and tries to carry on: refused. + kind, body = send(stalled, chunk(1)) + if kind != streamKindComplete { + t.Fatalf("stalled transfer's chunk answered with kind %#x, want a refusal", kind) + } + if ok, msg := decodeComplete(body); ok || !strings.Contains(msg, "taken it over") { + t.Fatalf("stalled transfer's chunk: ok=%v %q", ok, msg) + } + + for i := 1; i < 3; i++ { + if kind, _ := send(retry, chunk(i)); kind != streamKindAck { + t.Fatalf("retry chunk %d answered with kind %#x", i, kind) + } + } + kind, body = send(retry, encodeStreamFrame(streamKindDone, id, nil)) + if ok, msg := decodeComplete(body); kind != streamKindComplete || !ok { + t.Fatalf("retry DONE: kind %#x ok=%v %q", kind, ok, msg) + } + entries, _ := os.ReadDir(dir) + files := 0 + for _, e := range entries { + if e.IsDir() { + continue + } + files++ + got, _ := os.ReadFile(filepath.Join(dir, e.Name())) + if !bytes.Equal(got, data) { + t.Fatalf("%s does not match what was sent", e.Name()) + } + } + if files != 1 { + t.Fatalf("%d files received, want 1", files) + } + stalled.Close() + retry.Close() + if left := partialFiles(t, dir); len(left) > 0 { + t.Fatalf("partial files left behind: %v", left) + } +} From 099cf542ba08b7badac21dd5f0a80875e88c107f Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:44:51 +0300 Subject: [PATCH 3/6] filestream: document how SendFileStream reports a refusal at INIT A receiver that refuses a transfer at INIT answers with a COMPLETE, and SendFileStream returns that as a StreamResult with OK false and the reason in Message, not as ErrStreamUnsupported. The doc comment only described ErrStreamUnsupported. Co-Authored-By: Claude Opus 5.5 --- filestream.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/filestream.go b/filestream.go index 6af376c..435ec0f 100644 --- a/filestream.go +++ b/filestream.go @@ -195,8 +195,13 @@ type frameRW interface { // // Returns ErrStreamUnsupported if the peer never answers INIT with an // INIT-ACK within the negotiation window — the caller should fall back to -// SendFile on a fresh connection. stepTimeout bounds the wait for any single -// ACK and for the final COMPLETE (0 ⇒ default). +// SendFile on a fresh connection. A receiver that refuses the transfer at +// INIT (quota, disk space, too many transfers) answers with a COMPLETE +// instead; that comes back as a StreamResult with OK false and the reason in +// Message, and a nil error — not as ErrStreamUnsupported, since falling back +// would push the whole file at a peer that has just declined it. +// stepTimeout bounds the wait for any single ACK and for the final COMPLETE +// (0 ⇒ default). func (c *Client) SendFileStream(name string, r io.ReadSeeker, size int64, stepTimeout time.Duration) (*StreamResult, error) { return streamSend(c.conn, name, r, size, stepTimeout) } From 0ba6c7f7b5ef347c7e0b9fce209bad34c502708f Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:45:25 +0300 Subject: [PATCH 4/6] filestream: a .partial claim is taken over only by an accepted INIT, after two minutes Review of the takeover in the previous commit found these holes. - A second INIT on a connection whose transfer a retry had taken over reused that transfer's path and dead claim: it opened the retry's file, could truncate it or delete it on a disk-check refusal, and left the id refusing every chunk. Such a transfer is now dropped, without touching the file, and the INIT goes through the claim like a new one. - DONE checked the claim only on entry, then ran fsync, a hash over the whole file and onPrepare before renaming. It checks again right before the rename, and refreshes the claim while hashing. - An INIT refused for quota, disk space or too many transfers took the claim over first and then released it, so the quiet holder lost its file to nobody. The checks now run before the claim is taken, and a failure after it hands the claim back. The quota check also compares the declared size unsigned: one of 2^63 bytes or more came out negative and passed. - A claim went stale after 10s, while a sender waits up to 60s (pilotctl: 90s) for an ACK, so a live transfer on a slow link could lose its file. staleClaimAfter is now twice streamStepTimeout. A retry sooner than that gets a private file and starts from 0; the comment says so. - A DONE refused for a takeover forgot the transfer without closing its file. It and a chunk refused for a takeover now discard the transfer. - A transfer abandoned for a full disk released its claim before deleting the file, so it could delete a new holder's file. It deletes first, and only while it still holds the claim. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 +- filestream.go | 335 +++++++++++++++++------- zz_filestream_concurrent_test.go | 8 + zz_filestream_takeover_test.go | 419 +++++++++++++++++++++++++++++++ 4 files changed, 683 insertions(+), 89 deletions(-) create mode 100644 zz_filestream_takeover_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index bb3fdb4..0940cbb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,14 +37,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 rename. The first transfer keeps the resumable `.partial`; a concurrent one gets a private file that is removed if it does not finish. A sender that stalls and retries still resumes: once the transfer holding the `.partial` - has written nothing for 10 seconds, the retry takes the file over and the - stalled transfer is refused any further write. + has written nothing for two minutes (twice the sender's default wait for an + ACK), the retry takes the file over, and the stalled transfer is refused + any further write, its DONE, and the file itself if it starts over. A retry + sooner than that starts over in a private file. A transfer refused at the + start never takes a file over. - A file that cannot fit on the receiver's disk is refused before any byte is sent. The byte quota is a fixed number and can be larger than the disk, so a transfer ran until the disk was full, failed, and left its bytes in `.partial` — where they kept the disk full and blocked incoming messages. The receiver now checks free space at the start (keeping 16 MiB in reserve), and if a write still hits a full disk it deletes that `.partial`. +- A streamed transfer declaring a size of 2^63 bytes or more is refused by the + receiver's byte quota. The size was converted to a negative number and + passed the check. - A sender whose transfer is refused at the start (quota, disk full, too many transfers) reports the refusal. It was mistaken for a receiver too old to support streamed transfers, so the caller retried with the single-frame diff --git a/filestream.go b/filestream.go index 435ec0f..cbdc024 100644 --- a/filestream.go +++ b/filestream.go @@ -489,7 +489,10 @@ type recvTransfer struct { // over. That is a sender retrying after a stall: it reconnects while this // receiver still holds the stalled connection open (until the idle timeout, // minutes later). The retry takes the .partial over and resumes where the -// stalled transfer stopped; the stalled one is refused any further write. +// stalled transfer stopped; the stalled one is refused any further write, +// its DONE, and the file itself if it sends a new INIT. Only an INIT that is +// accepted takes a claim over: one refused for quota, disk space or too many +// transfers leaves the quiet holder its file. var activePartials = struct { sync.Mutex paths map[string]*partialClaim @@ -500,26 +503,60 @@ type partialClaim struct { lastWrite time.Time } +// live reports whether c's holder has written recently enough to keep it. +// Caller holds activePartials. +func (c *partialClaim) live() bool { + return time.Since(c.lastWrite) < staleClaimAfter +} + // staleClaimAfter is how long a claim's holder may go without writing before -// another transfer of the same content may take the claim over. A variable so -// tests can shorten it. -var staleClaimAfter = 10 * time.Second +// another transfer of the same content may take the claim over. +// +// It is twice streamStepTimeout, the sender's default wait for each ACK +// before it gives up (pilotctl send-file sets 90s). A live transfer can be +// quiet for most of that wait — retransmission backoff, a very slow link. +// Taking its file over then would fail a transfer that was still going, and +// its sender would report the file rejected even though the other transfer +// delivered the content. +// +// The cost is on the retry side. A retry that arrives within this window — +// a sender reconnecting quickly while this receiver still holds the +// half-open connection of its first attempt — finds the .partial held, gets +// a private file like any concurrent transfer of the same content, and +// starts over from offset 0. The first attempt's .partial stays on disk +// meanwhile and counts against the quota (near the quota, enough to get the +// retry refused) until its connection times out and a later retry resumes +// it. +// +// A variable so tests can shorten it. +var staleClaimAfter = 2 * streamStepTimeout -// claimPartial takes path for a new transfer. It returns nil when a live -// transfer holds it. -func claimPartial(path string) *partialClaim { +// partialHeld reports whether a live transfer holds path. +func partialHeld(path string) bool { activePartials.Lock() defer activePartials.Unlock() - if held, ok := activePartials.paths[path]; ok && time.Since(held.lastWrite) < staleClaimAfter { - return nil + held := activePartials.paths[path] + return held != nil && held.live() +} + +// claimPartial takes path for a new transfer, taking it over from a holder +// that has gone quiet; prev is that holder's claim (nil if path was free). +// It returns a nil claim when a live transfer holds path. +func claimPartial(path string) (c, prev *partialClaim) { + activePartials.Lock() + defer activePartials.Unlock() + prev = activePartials.paths[path] + if prev != nil && prev.live() { + return nil, nil } - c := &partialClaim{lastWrite: time.Now()} + c = &partialClaim{lastWrite: time.Now()} activePartials.paths[path] = c - return c + return c, prev } // stillHolds reports whether c is still the claim on path, and records a -// write by its holder when it is. +// write by its holder when it is. A claim just refreshed cannot be taken +// over for staleClaimAfter, so a holder may act on the file right after. func stillHolds(path string, c *partialClaim) bool { if c == nil { return true // a transfer that never needed a claim @@ -535,11 +572,32 @@ func stillHolds(path string, c *partialClaim) bool { // releasePartial gives up c's claim on path, if it still holds it. func releasePartial(path string, c *partialClaim) { + unclaimPartial(path, c, nil) +} + +// unclaimPartial gives up c's claim on path, if it still holds it, and hands +// the path back to prev, the claim c took over (nil: leave it free). +func unclaimPartial(path string, c, prev *partialClaim) { activePartials.Lock() - if activePartials.paths[path] == c { + defer activePartials.Unlock() + if activePartials.paths[path] != c { + return + } + if prev != nil { + activePartials.paths[path] = prev + } else { delete(activePartials.paths, path) } - activePartials.Unlock() +} + +// privatePartial names a .partial only one transfer uses, beside the +// content-addressed one at shared. +func privatePartial(shared string) (string, error) { + var suffix [4]byte + if _, err := rand.Read(suffix[:]); err != nil { + return "", err + } + return shared + "." + hex.EncodeToString(suffix[:]), nil } // release gives up t's claim on its .partial and, for a private one, removes @@ -667,31 +725,74 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame if err := os.MkdirAll(partialDir, 0700); err != nil { return encodeComplete(id, false, "mkdir partial: "+err.Error()) } - partial := filepath.Join(partialDir, hex.EncodeToString(id[:])) - private := false - var claim *partialClaim + shared := filepath.Join(partialDir, hex.EncodeToString(id[:])) + sr.mu.Lock() prior := sr.transfers[id] + if prior != nil && !stillHolds(prior.partial, prior.claim) { + // A second INIT for a transfer whose .partial a retry on another + // connection has since taken over. The file is the retry's now: + // drop this transfer without touching it, and start over as a new + // one. + delete(sr.transfers, id) + if prior.file != nil { + _ = prior.file.Close() + } + prior = nil + } + // A new id when already at capacity is refused before anything else: + // no claim taken, no descriptor opened. + full := prior == nil && len(sr.transfers) >= maxConcurrentTransfers sr.mu.Unlock() + if full { + return encodeComplete(id, false, "too many concurrent transfers") + } + + partial, private := shared, false + goPrivate := func() error { + p, err := privatePartial(shared) + partial, private = p, true + return err + } + var claim, prev *partialClaim if prior != nil { // A second INIT for a transfer this receiver already holds: keep - // its .partial (and the claim on it). + // its .partial (and the claim on it, refreshed just above). partial, private, claim = prior.partial, prior.private, prior.claim - } else if claim = claimPartial(partial); claim == nil { + } else if partialHeld(shared) { // Another live transfer is writing this content. Take a private // .partial rather than share the file with it. - var suffix [4]byte - if _, err := rand.Read(suffix[:]); err != nil { + if err := goPrivate(); err != nil { return encodeComplete(id, false, "partial name: "+err.Error()) } - partial += "." + hex.EncodeToString(suffix[:]) - private = true - claim = claimPartial(partial) } - // fail reports a refused INIT, giving back the claim just taken. + + // Everything that can refuse the transfer runs before the claim is + // taken: an INIT that is refused must not take the .partial over from a + // holder that has only gone quiet. + quotaBudget, refusal := sr.admit(partial, size) + if refusal != "" { + return encodeComplete(id, false, refusal) + } + if prior == nil { + if claim, prev = claimPartial(partial); claim == nil { + // A live transfer of the same content claimed the .partial + // since partialHeld looked. Go private after all; that file + // starts empty, so check again. + if err := goPrivate(); err != nil { + return encodeComplete(id, false, "partial name: "+err.Error()) + } + if quotaBudget, refusal = sr.admit(partial, size); refusal != "" { + return encodeComplete(id, false, refusal) + } + claim, prev = claimPartial(partial) + } + } + // fail reports an INIT refused after the claim was taken, handing the + // claim back to the transfer it was taken from. fail := func(reason string) *Frame { if prior == nil { - releasePartial(partial, claim) + unclaimPartial(partial, claim, prev) if private { _ = os.Remove(partial) } @@ -699,68 +800,32 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame return encodeComplete(id, false, reason) } - // Quota gate: reject up front if the declared size cannot fit alongside - // what is already on disk (excluding this transfer's own .partial, which - // resume credits back). Bounds disk use before a single chunk lands. - quotaBudget := int64(-1) - if sr.quotaBytes > 0 { - used := dirSizeExcluding(sr.receivedDir, partial) - budget := sr.quotaBytes - used - if budget < 0 { - budget = 0 - } - if int64(size) > budget { - return fail(fmt.Sprintf("receiver disk quota exceeded: need %d, %d of %d available", - size, budget, sr.quotaBytes)) - } - quotaBudget = budget - } - file, err := os.OpenFile(partial, os.O_RDWR|os.O_CREATE, 0600) if err != nil { return fail("open partial: " + err.Error()) } - // Resume from the contiguous bytes already on disk. Because we only - // ever write contiguously, the file size IS the resume offset. Guard - // against a stale/oversized .partial. - info, err := file.Stat() - var resume uint64 - if err == nil && info.Size() >= 0 && uint64(info.Size()) <= size { - resume = uint64(info.Size()) - } else { - _ = file.Truncate(0) - resume = 0 - } - - // Disk gate: the quota is a fixed number and can be larger than the - // disk. Refuse now what cannot fit, instead of failing mid-transfer with - // the disk full and the bytes already written stranded in a .partial. - if free, ok := freeDiskBytes(partialDir); ok { - need := size - resume - if need > free || free-need < diskReserveBytes { - _ = file.Close() - if resume == 0 { - _ = os.Remove(partial) - } - return fail(fmt.Sprintf("receiver disk full: need %d bytes, %d free (keeping %d in reserve)", - need, free, int64(diskReserveBytes))) - } - } sr.mu.Lock() - // Replace any stale in-memory transfer for this id (e.g. a prior conn). - if old := sr.transfers[id]; old != nil { - if old.file != nil { - _ = old.file.Close() - } - } else if len(sr.transfers) >= maxConcurrentTransfers { - // New id and already at capacity: refuse rather than opening - // another descriptor. Release the file we just opened, or this - // rejection would leak the very FD it exists to protect. + old := sr.transfers[id] + if old == nil && len(sr.transfers) >= maxConcurrentTransfers { + // Checked above, but another INIT can have taken the last slot + // since. Release the file we just opened, or this rejection would + // leak the very FD it exists to protect. sr.mu.Unlock() _ = file.Close() return fail("too many concurrent transfers") } + // Resume from the contiguous bytes already on disk, as admit counted + // on. A stale/oversized .partial starts over. + info, err := file.Stat() + resume := resumeOffset(info, err, size) + if resume == 0 { + _ = file.Truncate(0) + } + // Replace the in-memory transfer this INIT repeats, if any. + if old != nil && old.file != nil { + _ = old.file.Close() + } sr.transfers[id] = &recvTransfer{ file: file, partial: partial, @@ -778,6 +843,54 @@ func (sr *StreamReceiver) handleInit(id [transferIDLen]byte, body []byte) *Frame return encodeOffset(streamKindInitAck, id, resume) } +// admit runs the checks that can refuse a transfer of size bytes into +// partial, without touching the file. It returns the transfer's quota budget +// (-1 without a quota), or why the transfer is refused. +func (sr *StreamReceiver) admit(partial string, size uint64) (quotaBudget int64, refusal string) { + // Quota gate: reject up front if the declared size cannot fit alongside + // what is already on disk (excluding this transfer's own .partial, which + // resume credits back). Bounds disk use before a single chunk lands. + quotaBudget = -1 + if sr.quotaBytes > 0 { + used := dirSizeExcluding(sr.receivedDir, partial) + budget := sr.quotaBytes - used + if budget < 0 { + budget = 0 + } + // Compared unsigned: a declared size past the int64 range would + // otherwise come out negative and pass. + if size > uint64(budget) { + return 0, fmt.Sprintf("receiver disk quota exceeded: need %d, %d of %d available", + size, budget, sr.quotaBytes) + } + quotaBudget = budget + } + + // Disk gate: the quota is a fixed number and can be larger than the + // disk. Refuse now what cannot fit, instead of failing mid-transfer with + // the disk full and the bytes already written stranded in a .partial. + if free, ok := freeDiskBytes(filepath.Dir(partial)); ok { + info, err := os.Stat(partial) + need := size - resumeOffset(info, err, size) + if need > free || free-need < diskReserveBytes { + return 0, fmt.Sprintf("receiver disk full: need %d bytes, %d free (keeping %d in reserve)", + need, free, int64(diskReserveBytes)) + } + } + return quotaBudget, "" +} + +// resumeOffset is where a transfer of size bytes picks up in a .partial +// described by info (err from the stat). We only ever write contiguously, so +// the file size IS the resume offset — unless there is no file, or it is +// longer than the transfer and so stale: then it starts over at 0. +func resumeOffset(info os.FileInfo, err error, size uint64) uint64 { + if err == nil && info.Size() >= 0 && uint64(info.Size()) <= size { + return uint64(info.Size()) + } + return 0 +} + func (sr *StreamReceiver) handleChunk(id [transferIDLen]byte, body []byte) *Frame { off, data, ok := decodeChunk(body) if !ok { @@ -802,8 +915,7 @@ func (sr *StreamReceiver) handleChunk(id [transferIDLen]byte, body []byte) *Fram case off == t.cursor: if werr := sr.writeAt(t, off, data); werr != nil { sr.mu.Unlock() - sr.abandonIfDiskFull(id, werr) - return encodeComplete(id, false, werr.Error()) + return sr.writeFailed(id, werr) } // Drain any buffered successors. for { @@ -815,8 +927,7 @@ func (sr *StreamReceiver) handleChunk(id [transferIDLen]byte, body []byte) *Fram t.pendBytes -= len(next) if werr := sr.writeAt(t, t.cursor, next); werr != nil { sr.mu.Unlock() - sr.abandonIfDiskFull(id, werr) - return encodeComplete(id, false, werr.Error()) + return sr.writeFailed(id, werr) } } case off > t.cursor: @@ -855,6 +966,19 @@ func (sr *StreamReceiver) writeAt(t *recvTransfer, off uint64, data []byte) erro return nil } +// writeFailed answers a chunk whose write failed. A transfer whose file a +// retry has taken over can never write again, so it is ended here rather than +// holding its descriptor until the connection closes; one that ran the disk +// full is abandoned. +func (sr *StreamReceiver) writeFailed(id [transferIDLen]byte, werr error) *Frame { + if errors.Is(werr, errPartialTakenOver) { + sr.discard(id) + } else { + sr.abandonIfDiskFull(id, werr) + } + return encodeComplete(id, false, werr.Error()) +} + func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { sr.mu.Lock() t := sr.transfers[id] @@ -864,7 +988,7 @@ func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { } if !stillHolds(t.partial, t.claim) { // A retry of this transfer has taken the file over; it finishes it. - sr.forget(id) + sr.discard(id) return encodeComplete(id, false, errPartialTakenOver.Error()) } @@ -875,12 +999,18 @@ func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { return encodeComplete(id, false, fmt.Sprintf("incomplete: have %d of %d bytes", t.cursor, t.size)) } - // Verify the full content hash before accepting the file. + // Verify the full content hash before accepting the file. Reading + // through the claim keeps it fresh while a large file is hashed, and + // stops the hash if a retry has taken the file over all the same. if _, err := t.file.Seek(0, io.SeekStart); err != nil { return encodeComplete(id, false, "seek for verify: "+err.Error()) } h := sha256.New() - if _, err := io.Copy(h, t.file); err != nil { + if _, err := io.Copy(h, heldReader{t}); err != nil { + if errors.Is(err, errPartialTakenOver) { + sr.discard(id) + return encodeComplete(id, false, err.Error()) + } return encodeComplete(id, false, "read for verify: "+err.Error()) } if !bytes.Equal(h.Sum(nil), t.hash[:]) { @@ -898,6 +1028,14 @@ func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { return encodeComplete(id, false, "prepare: "+err.Error()) } } + // The claim again, right before the rename: fsync, the hash and + // onPrepare all take time, and a retry that took the file over + // meanwhile is writing it now. The check refreshes the claim, so it + // cannot be taken over between here and the rename. + if !stillHolds(t.partial, t.claim) { + sr.forget(id) + return encodeComplete(id, false, errPartialTakenOver.Error()) + } if err := os.Rename(t.partial, finalPath); err != nil { // Drop the in-memory transfer. t.file was already closed above, so // leaving the entry in place stranded a record holding a closed @@ -925,6 +1063,19 @@ func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { return encodeComplete(id, true, "") } +// heldReader reads t's .partial while t still holds its claim, refreshing the +// claim on every read so a long verification does not make the transfer look +// stalled. Once a retry has taken the file over it fails with +// errPartialTakenOver. +type heldReader struct{ t *recvTransfer } + +func (r heldReader) Read(p []byte) (int, error) { + if !stillHolds(r.t.partial, r.t.claim) { + return 0, errPartialTakenOver + } + return r.t.file.Read(p) +} + // discard closes and forgets a transfer but leaves the .partial on disk // (unless it is a private one, which nothing could resume). func (sr *StreamReceiver) discard(id [transferIDLen]byte) { @@ -960,11 +1111,21 @@ func (sr *StreamReceiver) abandonIfDiskFull(id [transferIDLen]byte, werr error) } sr.mu.Lock() t := sr.transfers[id] + delete(sr.transfers, id) sr.mu.Unlock() - sr.discard(id) - if t != nil { + if t == nil { + return + } + if t.file != nil { + _ = t.file.Close() + } + // Delete the file while the claim is still held, then release it: once + // released, another transfer can claim the path and create a new file + // there. A retry that has taken the file over owns it; leave it be. + if stillHolds(t.partial, t.claim) { _ = os.Remove(t.partial) } + t.release() } // Close releases any open .partial handles (call on connection teardown). diff --git a/zz_filestream_concurrent_test.go b/zz_filestream_concurrent_test.go index c80f3b2..b06abf7 100644 --- a/zz_filestream_concurrent_test.go +++ b/zz_filestream_concurrent_test.go @@ -288,6 +288,14 @@ func TestFileStream_RetryTakesOverAStalledTransfer(t *testing.T) { if ok, msg := decodeComplete(body); ok || !strings.Contains(msg, "taken it over") { t.Fatalf("stalled transfer's chunk: ok=%v %q", ok, msg) } + // It cannot write again, so it is ended rather than holding its file + // open until the connection closes. + stalled.mu.Lock() + _, kept := stalled.transfers[id] + stalled.mu.Unlock() + if kept { + t.Error("the stalled transfer was kept after its file was taken over") + } for i := 1; i < 3; i++ { if kind, _ := send(retry, chunk(i)); kind != streamKindAck { diff --git a/zz_filestream_takeover_test.go b/zz_filestream_takeover_test.go new file mode 100644 index 0000000..fe3468e --- /dev/null +++ b/zz_filestream_takeover_test.go @@ -0,0 +1,419 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +import ( + "bytes" + "crypto/sha256" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync/atomic" + "syscall" + "testing" + "time" +) + +// These tests drive StreamReceivers directly, one per connection, around the +// claim on a transfer's content-addressed .partial: which transfer may write +// it, and when a retry may take it over from one that has gone quiet. + +// shortenStaleClaim makes a claim go stale after d for the rest of the test. +func shortenStaleClaim(t *testing.T, d time.Duration) { + t.Helper() + prev := staleClaimAfter + staleClaimAfter = d + t.Cleanup(func() { staleClaimAfter = prev }) +} + +// setFreeDisk makes the receiver see free bytes of disk for the rest of the +// test. +func setFreeDisk(t *testing.T, free uint64) { + t.Helper() + prev := freeDiskBytes + freeDiskBytes = func(string) (uint64, bool) { return free, true } + t.Cleanup(func() { freeDiskBytes = prev }) +} + +// sequencedNames gives every received file its own name: the default one is +// timestamped to the millisecond, so two transfers finishing together would +// share it. +func sequencedNames() func(string) string { + var seq atomic.Int64 + return func(base string) string { return fmt.Sprintf("%d-%s", seq.Add(1), base) } +} + +// streamFile is a file sent in whole StreamChunkSize chunks. +type streamFile struct { + data []byte + id [transferIDLen]byte + hash [32]byte +} + +func newStreamFile(chunks int) streamFile { + data := makePayload(chunks * StreamChunkSize) + return streamFile{data: data, id: transferIDOf(data), hash: sha256.Sum256(data)} +} + +func (f streamFile) init() *Frame { return f.initSized(uint64(len(f.data))) } + +func (f streamFile) initSized(size uint64) *Frame { + return encodeInit(f.id, size, f.hash, StreamChunkSize, "f.bin") +} + +func (f streamFile) chunk(i int) *Frame { + return encodeChunk(f.id, uint64(i*StreamChunkSize), f.data[i*StreamChunkSize:(i+1)*StreamChunkSize]) +} + +func (f streamFile) done() *Frame { return encodeStreamFrame(streamKindDone, f.id, nil) } + +// exchange hands sr one frame and decodes its answer. +func exchange(t *testing.T, sr *StreamReceiver, f *Frame) (kind byte, body []byte) { + t.Helper() + resp := sr.HandleFrame(f) + if resp == nil { + t.Fatal("no response") + } + kind, _, body, ok := decodeStreamFrame(resp) + if !ok { + t.Fatal("malformed response") + } + return kind, body +} + +// expectResume sends an INIT and checks it is accepted at offset want. +func expectResume(t *testing.T, sr *StreamReceiver, f *Frame, want uint64) { + t.Helper() + kind, body := exchange(t, sr, f) + if kind != streamKindInitAck { + _, msg := decodeComplete(body) + t.Fatalf("INIT answered with kind %#x: %q", kind, msg) + } + if off, _ := decodeOffset(body); off != want { + t.Fatalf("INIT resumes at %d, want %d", off, want) + } +} + +// expectAck sends a chunk and checks it is written. +func expectAck(t *testing.T, sr *StreamReceiver, f *Frame) { + t.Helper() + if kind, body := exchange(t, sr, f); kind != streamKindAck { + _, msg := decodeComplete(body) + t.Fatalf("chunk answered with kind %#x: %q", kind, msg) + } +} + +// expectRefused sends f and checks it is refused for the reason want. +func expectRefused(t *testing.T, sr *StreamReceiver, f *Frame, want string) { + t.Helper() + kind, body := exchange(t, sr, f) + ok, msg := decodeComplete(body) + if kind != streamKindComplete || ok { + t.Fatalf("answered with kind %#x ok=%v, want a refusal", kind, ok) + } + if !strings.Contains(msg, want) { + t.Fatalf("refusal %q does not say %q", msg, want) + } +} + +// expectCompleted sends DONE and checks the file is accepted. +func expectCompleted(t *testing.T, sr *StreamReceiver, f streamFile) { + t.Helper() + kind, body := exchange(t, sr, f.done()) + if ok, msg := decodeComplete(body); kind != streamKindComplete || !ok { + t.Fatalf("DONE answered with kind %#x ok=%v %q", kind, ok, msg) + } +} + +// expectCopies checks dir holds n received files, each with data. +func expectCopies(t *testing.T, dir string, data []byte, n int) { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + files := 0 + for _, e := range entries { + if e.IsDir() { + continue + } + files++ + got, err := os.ReadFile(filepath.Join(dir, e.Name())) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, data) { + t.Errorf("%s does not match what was sent", e.Name()) + } + } + if files != n { + t.Fatalf("%d files received, want %d", files, n) + } +} + +// openTransferFile returns the descriptor sr holds for transfer id. +func openTransferFile(t *testing.T, sr *StreamReceiver, id [transferIDLen]byte) *os.File { + t.Helper() + sr.mu.Lock() + defer sr.mu.Unlock() + tr := sr.transfers[id] + if tr == nil { + t.Fatal("no transfer") + } + return tr.file +} + +// A stalled transfer whose .partial a retry has taken over is refused its +// DONE, and lets go of the file: it used to keep the descriptor open until +// its connection closed. +func TestFileStream_StalledDoneIsRefused(t *testing.T) { + shortenStaleClaim(t, 50*time.Millisecond) + dir := t.TempDir() + f := newStreamFile(2) + + stalled := NewStreamReceiver(dir, nil, nil) + defer stalled.Close() + expectResume(t, stalled, f.init(), 0) + expectAck(t, stalled, f.chunk(0)) + expectAck(t, stalled, f.chunk(1)) + time.Sleep(2 * staleClaimAfter) // the sender stalls before DONE, then retries + + retry := NewStreamReceiver(dir, nil, nil) + defer retry.Close() + expectResume(t, retry, f.init(), uint64(len(f.data))) + + file := openTransferFile(t, stalled, f.id) + expectRefused(t, stalled, f.done(), "taken it over") + if err := file.Close(); !errors.Is(err, os.ErrClosed) { + t.Error("the refused transfer's .partial is still open") + } + + expectCompleted(t, retry, f) + expectCopies(t, dir, f.data, 1) +} + +// DONE takes a while — fsync, a hash over the whole file, onPrepare — and a +// retry can take the .partial over meanwhile. The transfer that lost it must +// not then rename the file the retry is finishing. +func TestFileStream_TakeoverDuringDoneStopsTheRename(t *testing.T) { + shortenStaleClaim(t, 50*time.Millisecond) + dir := t.TempDir() + f := newStreamFile(2) + + retry := NewStreamReceiver(dir, nil, nil) + defer retry.Close() + onPrepare := func([transferIDLen]byte, string, string, int64) error { + time.Sleep(2 * staleClaimAfter) + expectResume(t, retry, f.init(), uint64(len(f.data))) + return nil + } + stalled := NewStreamReceiverWithQuotaAndPrepareAndCommit(dir, nil, nil, onPrepare, nil, 0) + defer stalled.Close() + expectResume(t, stalled, f.init(), 0) + expectAck(t, stalled, f.chunk(0)) + expectAck(t, stalled, f.chunk(1)) + expectRefused(t, stalled, f.done(), "taken it over") + + expectCompleted(t, retry, f) + expectCopies(t, dir, f.data, 1) +} + +// A second INIT on a connection whose transfer a retry has since taken over +// must leave the retry's file alone — not open it as its own, truncate it or +// delete it — and must not leave the transfer stuck refusing every chunk. +func TestFileStream_ReInitOnSupersededConnection(t *testing.T) { + // stall starts f on one connection, writes its first written chunks, and + // has a retry on another connection take the .partial over. + stall := func(t *testing.T, dir string, f streamFile, written int) (stalled, retry *StreamReceiver) { + t.Helper() + shortenStaleClaim(t, 50*time.Millisecond) + names := sequencedNames() + stalled = NewStreamReceiver(dir, names, nil) + t.Cleanup(stalled.Close) + expectResume(t, stalled, f.init(), 0) + for i := 0; i < written; i++ { + expectAck(t, stalled, f.chunk(i)) + } + time.Sleep(2 * staleClaimAfter) + retry = NewStreamReceiver(dir, names, nil) + t.Cleanup(retry.Close) + expectResume(t, retry, f.init(), uint64(written*StreamChunkSize)) + return stalled, retry + } + + t.Run("same file", func(t *testing.T) { + dir := t.TempDir() + f := newStreamFile(3) + stalled, retry := stall(t, dir, f, 1) + expectAck(t, retry, f.chunk(1)) + + // The stalled sender starts over on the connection it still has. + // The retry is writing the .partial, so this gets a private one. + expectResume(t, stalled, f.init(), 0) + for i := 0; i < 3; i++ { + expectAck(t, stalled, f.chunk(i)) + } + expectCompleted(t, stalled, f) + + expectAck(t, retry, f.chunk(2)) + expectCompleted(t, retry, f) + expectCopies(t, dir, f.data, 2) + stalled.Close() + retry.Close() + if left := partialFiles(t, dir); len(left) > 0 { + t.Fatalf("partial files left behind: %v", left) + } + }) + + t.Run("declares less than the file holds", func(t *testing.T) { + dir := t.TempDir() + f := newStreamFile(3) + stalled, retry := stall(t, dir, f, 1) + expectAck(t, retry, f.chunk(1)) + + // A .partial longer than the declared size is truncated. + stalled.HandleFrame(f.initSized(StreamChunkSize)) + + expectAck(t, retry, f.chunk(2)) + expectCompleted(t, retry, f) + expectCopies(t, dir, f.data, 1) + }) + + t.Run("refused by the disk check", func(t *testing.T) { + dir := t.TempDir() + f := newStreamFile(2) + stalled, retry := stall(t, dir, f, 0) + + // A refusal used to delete an empty .partial on its way out. + setFreeDisk(t, 0) + expectRefused(t, stalled, f.init(), "disk full") + + expectAck(t, retry, f.chunk(0)) + expectAck(t, retry, f.chunk(1)) + expectCompleted(t, retry, f) + expectCopies(t, dir, f.data, 1) + }) +} + +// An INIT that is refused must not take a .partial over from a transfer that +// has merely gone quiet. Anyone who knows the transfer id could otherwise +// strip a quiet transfer of its file by declaring an absurd size. +func TestFileStream_RefusedInitDoesNotTakeOver(t *testing.T) { + for _, tc := range []struct { + name string + refuse func(t *testing.T, dir string, f streamFile) + }{ + {"quota", func(t *testing.T, dir string, f streamFile) { + intruder := NewStreamReceiverWithQuota(dir, nil, nil, 1<<20) + defer intruder.Close() + expectRefused(t, intruder, f.initSized(1<<40), "quota exceeded") + }}, + {"quota, a size past the int64 range", func(t *testing.T, dir string, f streamFile) { + // Where free space cannot be read, the quota is the only gate. + prev := freeDiskBytes + freeDiskBytes = func(string) (uint64, bool) { return 0, false } + t.Cleanup(func() { freeDiskBytes = prev }) + intruder := NewStreamReceiverWithQuota(dir, nil, nil, 1<<20) + defer intruder.Close() + expectRefused(t, intruder, f.initSized(1<<63), "quota exceeded") + }}, + {"disk space", func(t *testing.T, dir string, f streamFile) { + setFreeDisk(t, 1<<30) + intruder := NewStreamReceiver(dir, nil, nil) + defer intruder.Close() + expectRefused(t, intruder, f.initSized(1<<40), "disk full") + }}, + {"too many transfers", func(t *testing.T, dir string, f streamFile) { + intruder := NewStreamReceiver(dir, nil, nil) + defer intruder.Close() + for i := 0; i < maxConcurrentTransfers; i++ { + other := f.id + other[0], other[1] = ^other[0], byte(i) + expectResume(t, intruder, encodeInit(other, 1, f.hash, StreamChunkSize, "x"), 0) + } + expectRefused(t, intruder, f.init(), "too many concurrent transfers") + }}, + } { + t.Run(tc.name, func(t *testing.T) { + shortenStaleClaim(t, 50*time.Millisecond) + dir := t.TempDir() + f := newStreamFile(2) + + holder := NewStreamReceiver(dir, nil, nil) + defer holder.Close() + expectResume(t, holder, f.init(), 0) + expectAck(t, holder, f.chunk(0)) + time.Sleep(2 * staleClaimAfter) + + tc.refuse(t, dir, f) + + expectAck(t, holder, f.chunk(1)) + expectCompleted(t, holder, f) + expectCopies(t, dir, f.data, 1) + }) + } +} + +// What makes a claim stale is how long its holder has gone without writing, +// not how old the claim is. +func TestFileStream_LiveHolderIsNotTakenOver(t *testing.T) { + shortenStaleClaim(t, 100*time.Millisecond) + dir := t.TempDir() + f := newStreamFile(2) + names := sequencedNames() + + holder := NewStreamReceiver(dir, names, nil) + defer holder.Close() + expectResume(t, holder, f.init(), 0) + time.Sleep(2 * staleClaimAfter) + expectAck(t, holder, f.chunk(0)) + + // The same content on another connection: a private file, from 0. + other := NewStreamReceiver(dir, names, nil) + defer other.Close() + expectResume(t, other, f.init(), 0) + + expectAck(t, holder, f.chunk(1)) + expectCompleted(t, holder, f) + expectAck(t, other, f.chunk(0)) + expectAck(t, other, f.chunk(1)) + expectCompleted(t, other, f) + expectCopies(t, dir, f.data, 2) +} + +// A sender waits up to streamStepTimeout for each ACK, so a live transfer can +// be quiet that long. Its claim must not go stale sooner: losing the file +// fails a transfer that was still going. +func TestFileStream_StaleClaimOutlastsSenderStepTimeout(t *testing.T) { + if staleClaimAfter <= streamStepTimeout { + t.Fatalf("a claim goes stale after %s, but a live sender waits up to %s for an ACK", + staleClaimAfter, streamStepTimeout) + } +} + +// A transfer abandoned for a full disk deletes its .partial only while it +// still holds it: once a retry has taken the file over, the file is the +// retry's. +func TestFileStream_DiskFullLeavesATakenOverFileAlone(t *testing.T) { + shortenStaleClaim(t, 50*time.Millisecond) + dir := t.TempDir() + f := newStreamFile(2) + + stalled := NewStreamReceiver(dir, nil, nil) + defer stalled.Close() + expectResume(t, stalled, f.init(), 0) + expectAck(t, stalled, f.chunk(0)) + time.Sleep(2 * staleClaimAfter) + + retry := NewStreamReceiver(dir, nil, nil) + defer retry.Close() + expectResume(t, retry, f.init(), StreamChunkSize) + + stalled.abandonIfDiskFull(f.id, &os.PathError{Op: "write", Path: "f", Err: syscall.ENOSPC}) + + expectAck(t, retry, f.chunk(1)) + expectCompleted(t, retry, f) + expectCopies(t, dir, f.data, 1) +} From 856f7a6eb2173176e2f2844edec2fb1ba2626e7f Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:45:39 +0300 Subject: [PATCH 5/6] filestream: recognize a full disk per platform; plan9 builds again syscall.ENOSPC does not exist on plan9, so the disk-full check broke the plan9 build. The errors that mean a full disk now live behind a build-tag split: ENOSPC and EDQUOT (a full disk quota) everywhere but Windows and Plan 9, ERROR_DISK_FULL and ERROR_HANDLE_DISK_FULL on Windows, none on Plan 9, whose errors are strings. FreeBSD's Statfs_t.Bavail is signed and goes negative once root's reserved blocks are in use; that now counts as no free space rather than converting to an enormous number. A test drives a disk-full write through handleChunk, for each error the platform recognizes, using a writePartialAt seam. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 3 ++- filestream.go | 20 +++++++++++--- filestream_disk_unix.go | 12 ++++++++- filestream_diskfull.go | 11 ++++++++ filestream_diskfull_plan9.go | 8 ++++++ filestream_diskfull_windows.go | 12 +++++++++ zz_filestream_concurrent_test.go | 45 ++++++++++++++++++++++++++++++-- zz_filestream_disk_unix_test.go | 21 +++++++++++++++ zz_filestream_takeover_test.go | 6 +++-- 9 files changed, 129 insertions(+), 9 deletions(-) create mode 100644 filestream_diskfull.go create mode 100644 filestream_diskfull_plan9.go create mode 100644 filestream_diskfull_windows.go create mode 100644 zz_filestream_disk_unix_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 0940cbb..6da8acc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,7 +47,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 transfer ran until the disk was full, failed, and left its bytes in `.partial` — where they kept the disk full and blocked incoming messages. The receiver now checks free space at the start (keeping 16 MiB in reserve), - and if a write still hits a full disk it deletes that `.partial`. + and if a write still hits a full disk or disk quota it deletes that + `.partial`. - A streamed transfer declaring a size of 2^63 bytes or more is refused by the receiver's byte quota. The size was converted to a negative number and passed the check. diff --git a/filestream.go b/filestream.go index cbdc024..4b3f872 100644 --- a/filestream.go +++ b/filestream.go @@ -39,7 +39,6 @@ import ( "path/filepath" "sync" "sync/atomic" - "syscall" "time" ) @@ -959,13 +958,17 @@ func (sr *StreamReceiver) writeAt(t *recvTransfer, off uint64, data []byte) erro } t.quotaBudget -= int64(len(data)) } - if _, err := t.file.WriteAt(data, int64(off)); err != nil { + if _, err := writePartialAt(t.file, data, int64(off)); err != nil { return fmt.Errorf("write at %d: %w", off, err) } t.cursor += uint64(len(data)) return nil } +// writePartialAt writes a chunk into a .partial. A variable so tests can make +// the write fail. +var writePartialAt = (*os.File).WriteAt + // writeFailed answers a chunk whose write failed. A transfer whose file a // retry has taken over can never write again, so it is ended here rather than // holding its descriptor until the connection closes; one that ran the disk @@ -1106,7 +1109,7 @@ func (sr *StreamReceiver) forget(id [transferIDLen]byte) { // could then store neither messages nor its own state — for the sake of a // resume that cannot succeed until space is freed anyway. func (sr *StreamReceiver) abandonIfDiskFull(id [transferIDLen]byte, werr error) { - if !errors.Is(werr, syscall.ENOSPC) { + if !isDiskFull(werr) { return } sr.mu.Lock() @@ -1128,6 +1131,17 @@ func (sr *StreamReceiver) abandonIfDiskFull(id [transferIDLen]byte, werr error) t.release() } +// isDiskFull reports whether err is a write failing for lack of space: +// one of the platform's diskFullErrors. +func isDiskFull(err error) bool { + for _, target := range diskFullErrors { + if errors.Is(err, target) { + return true + } + } + return false +} + // Close releases any open .partial handles (call on connection teardown). // The .partial files themselves remain for resume, except private ones. func (sr *StreamReceiver) Close() { diff --git a/filestream_disk_unix.go b/filestream_disk_unix.go index 6545a78..f554219 100644 --- a/filestream_disk_unix.go +++ b/filestream_disk_unix.go @@ -15,5 +15,15 @@ func platformFreeDiskBytes(dir string) (uint64, bool) { if err := syscall.Statfs(dir, &st); err != nil { return 0, false } - return uint64(st.Bavail) * uint64(st.Bsize), true // #nosec G115 -- block size is positive + return blocksAvailable(st.Bavail) * uint64(st.Bsize), true // #nosec G115 -- block size is positive +} + +// blocksAvailable converts Statfs_t.Bavail to a block count. It is signed on +// FreeBSD, and negative there once the blocks reserved for root are in use: +// nothing is available. +func blocksAvailable[T int64 | uint64](n T) uint64 { + if n < 0 { + return 0 + } + return uint64(n) // #nosec G115 -- not negative } diff --git a/filestream_diskfull.go b/filestream_diskfull.go new file mode 100644 index 0000000..327605f --- /dev/null +++ b/filestream_diskfull.go @@ -0,0 +1,11 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !windows && !plan9 + +package dataexchange + +import "syscall" + +// diskFullErrors are what a write fails with when the filesystem has no room +// for it: out of space, or out of the user's disk quota. +var diskFullErrors = []error{syscall.ENOSPC, syscall.EDQUOT} diff --git a/filestream_diskfull_plan9.go b/filestream_diskfull_plan9.go new file mode 100644 index 0000000..23ed940 --- /dev/null +++ b/filestream_diskfull_plan9.go @@ -0,0 +1,8 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +// diskFullErrors is empty: Plan 9 reports errors as strings, with no code to +// match a full disk by. A write that fails for lack of space keeps its +// .partial, like any other failed write. +var diskFullErrors []error diff --git a/filestream_diskfull_windows.go b/filestream_diskfull_windows.go new file mode 100644 index 0000000..241ae61 --- /dev/null +++ b/filestream_diskfull_windows.go @@ -0,0 +1,12 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +import "syscall" + +// diskFullErrors are what a write fails with when the volume has no room for +// it. syscall has no names for these Windows error codes. +var diskFullErrors = []error{ + syscall.Errno(112), // ERROR_DISK_FULL + syscall.Errno(39), // ERROR_HANDLE_DISK_FULL +} diff --git a/zz_filestream_concurrent_test.go b/zz_filestream_concurrent_test.go index b06abf7..5756c1e 100644 --- a/zz_filestream_concurrent_test.go +++ b/zz_filestream_concurrent_test.go @@ -12,7 +12,6 @@ import ( "strings" "sync" "sync/atomic" - "syscall" "testing" "time" ) @@ -153,6 +152,9 @@ func TestFileStream_RefusedWhenDiskCannotHoldIt(t *testing.T) { // When a write fails because the disk is full, the .partial goes: keeping it // holds the disk full for a resume that cannot succeed. func TestFileStream_DiskFullMidTransferRemovesPartial(t *testing.T) { + if len(diskFullErrors) == 0 { + t.Skip("no disk-full error is recognized on this platform") + } dir := t.TempDir() data := makePayload(StreamChunkSize) id := transferIDOf(data) @@ -185,7 +187,7 @@ func TestFileStream_DiskFullMidTransferRemovesPartial(t *testing.T) { if len(partialFiles(t, dir)) != 1 { t.Fatal("a non-disk-full error must keep the partial for resume") } - sr.abandonIfDiskFull(id, fmt.Errorf("write at 0: %w", syscall.ENOSPC)) + sr.abandonIfDiskFull(id, fmt.Errorf("write at 0: %w", diskFullErrors[0])) if left := partialFiles(t, dir); len(left) != 0 { t.Errorf("partial kept after disk-full: %v", left) } @@ -193,6 +195,45 @@ func TestFileStream_DiskFullMidTransferRemovesPartial(t *testing.T) { _ = srv.Close() } +// The same through the receiver's own write path, for each error that means +// the disk is full: the chunk is refused, the .partial deleted, the transfer +// ended and its claim given up. +func TestFileStream_DiskFullChunkRemovesPartial(t *testing.T) { + if len(diskFullErrors) == 0 { + t.Skip("no disk-full error is recognized on this platform") + } + for _, full := range diskFullErrors { + t.Run(full.Error(), func(t *testing.T) { + dir := t.TempDir() + f := newStreamFile(2) + sr := NewStreamReceiver(dir, nil, nil) + defer sr.Close() + expectResume(t, sr, f.init(), 0) + expectAck(t, sr, f.chunk(0)) + + prev := writePartialAt + writePartialAt = func(file *os.File, _ []byte, _ int64) (int, error) { + return 0, &os.PathError{Op: "write", Path: file.Name(), Err: full} + } + defer func() { writePartialAt = prev }() + expectRefused(t, sr, f.chunk(1), "write at") + + if left := partialFiles(t, dir); len(left) != 0 { + t.Errorf("partial kept after disk-full: %v", left) + } + sr.mu.Lock() + _, kept := sr.transfers[f.id] + sr.mu.Unlock() + if kept { + t.Error("the transfer was kept after the disk filled") + } + if partialHeld(filepath.Join(dir, ".partial", fmt.Sprintf("%x", f.id))) { + t.Error("the abandoned transfer still holds its .partial") + } + }) + } +} + type writeCounter struct { bytes.Buffer writes int diff --git a/zz_filestream_disk_unix_test.go b/zz_filestream_disk_unix_test.go new file mode 100644 index 0000000..cb09ff1 --- /dev/null +++ b/zz_filestream_disk_unix_test.go @@ -0,0 +1,21 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build linux || darwin || freebsd + +package dataexchange + +import "testing" + +// FreeBSD's Statfs_t.Bavail is signed and goes negative once root's reserved +// blocks are in use. That is no space, not 2^64 blocks of it. +func TestBlocksAvailable_NegativeIsNone(t *testing.T) { + if got := blocksAvailable(int64(-3)); got != 0 { + t.Errorf("blocksAvailable(-3) = %d, want 0", got) + } + if got := blocksAvailable(int64(7)); got != 7 { + t.Errorf("blocksAvailable(int64 7) = %d, want 7", got) + } + if got := blocksAvailable(uint64(7)); got != 7 { + t.Errorf("blocksAvailable(uint64 7) = %d, want 7", got) + } +} diff --git a/zz_filestream_takeover_test.go b/zz_filestream_takeover_test.go index fe3468e..75ef5f5 100644 --- a/zz_filestream_takeover_test.go +++ b/zz_filestream_takeover_test.go @@ -11,7 +11,6 @@ import ( "path/filepath" "strings" "sync/atomic" - "syscall" "testing" "time" ) @@ -397,6 +396,9 @@ func TestFileStream_StaleClaimOutlastsSenderStepTimeout(t *testing.T) { // still holds it: once a retry has taken the file over, the file is the // retry's. func TestFileStream_DiskFullLeavesATakenOverFileAlone(t *testing.T) { + if len(diskFullErrors) == 0 { + t.Skip("no disk-full error is recognized on this platform") + } shortenStaleClaim(t, 50*time.Millisecond) dir := t.TempDir() f := newStreamFile(2) @@ -411,7 +413,7 @@ func TestFileStream_DiskFullLeavesATakenOverFileAlone(t *testing.T) { defer retry.Close() expectResume(t, retry, f.init(), StreamChunkSize) - stalled.abandonIfDiskFull(f.id, &os.PathError{Op: "write", Path: "f", Err: syscall.ENOSPC}) + stalled.abandonIfDiskFull(f.id, &os.PathError{Op: "write", Path: "f", Err: diskFullErrors[0]}) expectAck(t, retry, f.chunk(1)) expectCompleted(t, retry, f) From ea6fb65abc8b6d95769ab3cde3a249d34a887b57 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 17:04:26 +0300 Subject: [PATCH 6/6] test: the disk-full test waits for the transfer to be registered INIT creates the .partial before it records the transfer, so under load the test could abandon a transfer that was not recorded yet, which does nothing, and fail with "partial kept after disk-full". Review saw 4 of 12 stressed runs fail. Co-Authored-By: Claude Opus 5.5 --- zz_filestream_concurrent_test.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/zz_filestream_concurrent_test.go b/zz_filestream_concurrent_test.go index 5756c1e..9e53539 100644 --- a/zz_filestream_concurrent_test.go +++ b/zz_filestream_concurrent_test.go @@ -175,12 +175,21 @@ func TestFileStream_DiskFullMidTransferRemovesPartial(t *testing.T) { } }() go func() { _, _ = streamSend(cli, "f.bin", bytes.NewReader(data), int64(len(data)), time.Second) }() + // Wait for the transfer to be registered, not just for its file: INIT + // creates the .partial before it records the transfer, and abandoning + // an unrecorded transfer does nothing. + registered := func() bool { + sr.mu.Lock() + defer sr.mu.Unlock() + _, ok := sr.transfers[id] + return ok + } deadline := time.Now().Add(2 * time.Second) - for len(partialFiles(t, dir)) == 0 && time.Now().Before(deadline) { + for !registered() && time.Now().Before(deadline) { time.Sleep(5 * time.Millisecond) } - if len(partialFiles(t, dir)) != 1 { - t.Fatalf("expected one partial after INIT, have %v", partialFiles(t, dir)) + if len(partialFiles(t, dir)) != 1 || !registered() { + t.Fatalf("expected one registered partial after INIT, have %v", partialFiles(t, dir)) } sr.abandonIfDiskFull(id, fmt.Errorf("write at 0: %w", os.ErrPermission))