diff --git a/CHANGELOG.md b/CHANGELOG.md index 06841b3..6da8acc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,34 @@ 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 sender that + stalls and retries still resumes: once the transfer holding the `.partial` + 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 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. +- 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..4b3f872 100644 --- a/filestream.go +++ b/filestream.go @@ -28,6 +28,7 @@ package dataexchange import ( "bytes" + "crypto/rand" "crypto/sha256" "encoding/binary" "encoding/hex" @@ -193,8 +194,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) } @@ -248,6 +254,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 +452,16 @@ 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 + // 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 @@ -448,6 +475,152 @@ 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. +// +// 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, +// 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 +}{paths: make(map[string]*partialClaim)} + +// partialClaim is one transfer's hold on a .partial path. +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. +// +// 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 + +// partialHeld reports whether a live transfer holds path. +func partialHeld(path string) bool { + activePartials.Lock() + defer activePartials.Unlock() + 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()} + activePartials.paths[path] = c + return c, prev +} + +// stillHolds reports whether c is still the claim on path, and records a +// 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 + } + activePartials.Lock() + defer activePartials.Unlock() + if activePartials.paths[path] != c { + return false + } + c.lastWrite = time.Now() + return true +} + +// 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() + defer activePartials.Unlock() + if activePartials.paths[path] != c { + return + } + if prev != nil { + activePartials.paths[path] = prev + } else { + delete(activePartials.paths, path) + } +} + +// 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 +// the file (a no-op once a finished transfer has renamed it into place). +func (t *recvTransfer) release() { + 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. +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 @@ -551,59 +724,112 @@ 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[:])) + shared := filepath.Join(partialDir, hex.EncodeToString(id[:])) - // 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 + 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() } - if int64(size) > budget { - return encodeComplete(id, false, - fmt.Sprintf("receiver disk quota exceeded: need %d, %d of %d available", - size, budget, sr.quotaBytes)) + 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, refreshed just above). + partial, private, claim = prior.partial, prior.private, prior.claim + } else if partialHeld(shared) { + // Another live transfer is writing this content. Take a private + // .partial rather than share the file with it. + if err := goPrivate(); err != nil { + return encodeComplete(id, false, "partial name: "+err.Error()) } - quotaBudget = budget + } + + // 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 { + unclaimPartial(partial, claim, prev) + if private { + _ = os.Remove(partial) + } + } + return encodeComplete(id, false, reason) } file, err := os.OpenFile(partial, os.O_RDWR|os.O_CREATE, 0600) if err != nil { - return encodeComplete(id, false, "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 + return fail("open partial: " + err.Error()) } 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 encodeComplete(id, false, "too many concurrent transfers") + 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, + private: private, + claim: claim, name: sanitizeBase(name), size: size, hash: hash, @@ -616,6 +842,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 { @@ -640,7 +914,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() - return encodeComplete(id, false, werr.Error()) + return sr.writeFailed(id, werr) } // Drain any buffered successors. for { @@ -652,7 +926,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() - return encodeComplete(id, false, werr.Error()) + return sr.writeFailed(id, werr) } } case off > t.cursor: @@ -675,19 +949,39 @@ 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) } 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 +// 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] @@ -695,6 +989,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.discard(id) + return encodeComplete(id, false, errPartialTakenOver.Error()) + } if err := t.file.Sync(); err != nil { return encodeComplete(id, false, "fsync: "+err.Error()) @@ -703,12 +1002,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[:]) { @@ -726,6 +1031,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 @@ -753,33 +1066,95 @@ func (sr *StreamReceiver) handleDone(id [transferIDLen]byte) *Frame { return encodeComplete(id, true, "") } -// discard closes and forgets a transfer but leaves the .partial on disk. +// 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) { 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 !isDiskFull(werr) { + return + } + sr.mu.Lock() + t := sr.transfers[id] + delete(sr.transfers, id) + sr.mu.Unlock() + 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() +} + +// 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. +// 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..194e945 --- /dev/null +++ b/filestream_disk_other.go @@ -0,0 +1,9 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !linux && !darwin && !freebsd + +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..f554219 --- /dev/null +++ b/filestream_disk_unix.go @@ -0,0 +1,29 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build linux || darwin || freebsd + +package dataexchange + +import "syscall" + +// platformFreeDiskBytes reports the space available to this process on the +// 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 { + return 0, false + } + 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 new file mode 100644 index 0000000..9e53539 --- /dev/null +++ b/zz_filestream_concurrent_test.go @@ -0,0 +1,379 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +import ( + "bytes" + "crypto/sha256" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "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) { + if len(diskFullErrors) == 0 { + t.Skip("no disk-full error is recognized on this platform") + } + 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) }() + // 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 !registered() && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + 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)) + 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", diskFullErrors[0])) + if left := partialFiles(t, dir); len(left) != 0 { + t.Errorf("partial kept after disk-full: %v", left) + } + _ = cli.Close() + _ = 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 +} + +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) + } + } +} + +// 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) + } + // 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 { + 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) + } +} 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 new file mode 100644 index 0000000..75ef5f5 --- /dev/null +++ b/zz_filestream_takeover_test.go @@ -0,0 +1,421 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package dataexchange + +import ( + "bytes" + "crypto/sha256" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync/atomic" + "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) { + 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) + + 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: diskFullErrors[0]}) + + expectAck(t, retry, f.chunk(1)) + expectCompleted(t, retry, f) + expectCopies(t, dir, f.data, 1) +} 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") }