From 8f8774ea803ab7def6e7ac4f6af33473b1485816 Mon Sep 17 00:00:00 2001 From: go-while Date: Thu, 12 Nov 2026 13:43:33 +0000 Subject: [PATCH 1/2] fakeserver: CompressShape shapes the COMPRESS DEFLATE output stream Flush kind (sync, zlib-style partial flush like INN's nnrpd, final block with trailing bytes), compressed bytes split across socket writes (offsets from the start or end, so also inside the 00 00 FF FF marker), an empty flush or pipelined payload in the same socket write as the 206. Option on StoreOptions, PeerOptions, Step and Conn.StartCompressShaped; the zero value is the old behaviour. Co-Authored-By: Claude Opus 5.5 --- fakeserver/compress.go | 110 ++++++-- fakeserver/compressshape.go | 201 ++++++++++++++ fakeserver/compressshape_test.go | 435 +++++++++++++++++++++++++++++++ fakeserver/conn.go | 13 +- fakeserver/doc.go | 23 +- fakeserver/peer.go | 4 + fakeserver/server.go | 17 +- fakeserver/storehandler.go | 1 + 8 files changed, 771 insertions(+), 33 deletions(-) create mode 100644 fakeserver/compressshape.go create mode 100644 fakeserver/compressshape_test.go diff --git a/fakeserver/compress.go b/fakeserver/compress.go index 50bc11b..7994231 100644 --- a/fakeserver/compress.go +++ b/fakeserver/compress.go @@ -2,13 +2,13 @@ package fakeserver import ( "bufio" - "bytes" - "compress/flate" "errors" "io" "net" "regexp" + "slices" "strings" + "time" "github.com/go-while/nntp-testkit/internal/inflate" ) @@ -27,7 +27,12 @@ import ( // handlers return and the connection is closed, as RFC 8054 2.2.2 requires. Raw returns the // compressing conn afterwards; keep Raw() from before the call to write bytes past the // compressor (corrupt output for a client test). It fails when compression is already on. -func (c *Conn) StartCompress() error { return c.startCompress(false) } +func (c *Conn) StartCompress() error { return c.startCompress(false, CompressShape{}) } + +// StartCompressShaped is StartCompress with the output stream shaped (CompressShape): flush +// kind, splits, a final block. EmptyFlushWith206 and With206 have no effect here, since the 206 +// is already written; the handlers of the package write them with the 206. +func (c *Conn) StartCompressShaped(shape CompressShape) error { return c.startCompress(false, shape) } // StartCompressOutputOnly is a broken COMPRESS for client tests: after the 206 only what the // server writes is deflated, while the client's bytes are read as they arrive, without @@ -35,7 +40,7 @@ func (c *Conn) StartCompress() error { return c.startCompress(false) } // input. A client that compresses its commands, as RFC 8054 requires, then gets no answer // (its deflated bytes rarely hold a line end) or "500" answers to binary lines, and Lines // records those raw bytes. -func (c *Conn) StartCompressOutputOnly() error { return c.startCompress(true) } +func (c *Conn) StartCompressOutputOnly() error { return c.startCompress(true, CompressShape{}) } // Compressed reports whether StartCompress or StartCompressOutputOnly switched the connection // to compression. @@ -44,16 +49,19 @@ func (c *Conn) Compressed() bool { return ok } -func (c *Conn) startCompress(outputOnly bool) error { +func (c *Conn) startCompress(outputOnly bool, shape CompressShape) error { if c.Compressed() { return errors.New("fakeserver: compression is already active") } - d := &deflateConn{Conn: c.conn, r: c.r} - zw, err := flate.NewWriter(&d.out, flate.DefaultCompression) - if err != nil { - return err + enc := c.enc // the one that wrote the pipelined bytes of the 206, when it did + c.enc = nil + if enc == nil { + var err error + if enc, err = newEncoder(shape); err != nil { + return err + } } - d.zw = zw + d := &deflateConn{Conn: c.conn, r: c.r, enc: enc, sleep: c.sleep, deadline: c.s.deadline} if !outputOnly { pr, pw := io.Pipe() d.r, d.pr = pr, pr @@ -74,11 +82,12 @@ func (c *Conn) startCompress(outputOnly bool) error { // addresses and Close are those of the conn underneath. type deflateConn struct { net.Conn - r io.Reader - pr *io.PipeReader // the inflater's output; nil for StartCompressOutputOnly - zw *flate.Writer // writes into out - out bytes.Buffer // the deflated bytes of one Write, sent in one write - werr error // the first failed write: the stream is broken from there on + r io.Reader + pr *io.PipeReader // the inflater's output; nil for StartCompressOutputOnly + enc *encoder + sleep func(time.Duration) bool // Conn.sleep: bounded by the Deadline, false when the server closed + deadline time.Duration + werr error // the first failed write: the stream is broken from there on } func (d *deflateConn) Read(p []byte) (int, error) { return d.r.Read(p) } @@ -91,29 +100,56 @@ func (d *deflateConn) Close() error { return d.Conn.Close() } -// Write deflates p, flushes the compressor (sync flush) and writes the result to the conn -// underneath in one write, as a server that flushes its output buffer, so the client can -// inflate all of p at once (compress/flate's Writer on its own writes a short input in five -// pieces). After a failed write the deflate stream is broken, so every later Write returns -// that error. +// Write deflates p, flushes the compressor as the shape says (sync flush by default) and +// writes the result to the conn underneath in one write (or the pieces of the shape's Split), +// as a server that flushes its output buffer, so the client can inflate all of p at once +// (compress/flate's Writer on its own writes a short input in five pieces). After a failed +// write the deflate stream is broken, so every later Write returns that error. func (d *deflateConn) Write(p []byte) (int, error) { if d.werr != nil { return 0, d.werr } - d.out.Reset() - if _, err := d.zw.Write(p); err != nil { // into d.out: cannot fail - return 0, err - } - if err := d.zw.Flush(); err != nil { + b, err := d.enc.write(p) + if err != nil { return 0, err } - if _, err := d.Conn.Write(d.out.Bytes()); err != nil { + if err := d.send(b); err != nil { d.werr = err return 0, err } return len(p), nil } +// send writes b to the conn underneath, cut by the shape's Split. The pauses between the +// pieces end at the server's Deadline from the call, with a write deadline set meanwhile. +func (d *deflateConn) send(b []byte) error { + if len(b) == 0 { + return nil + } + sp := d.enc.shape.Split + cuts := sp.cuts(len(b)) + if len(cuts) == 0 { + _, err := d.Conn.Write(b) + return err + } + end := time.Now().Add(d.deadline) + if sp.Gap > 0 { + _ = d.Conn.SetWriteDeadline(end) + defer d.Conn.SetWriteDeadline(time.Time{}) + } + prev := 0 + for _, k := range append(cuts, len(b)) { + if _, err := d.Conn.Write(b[prev:k]); err != nil { + return err + } + prev = k + if k < len(b) && sp.Gap > 0 && (time.Now().After(end) || !d.sleep(sp.Gap)) { + return errors.New("fakeserver: compressed split ran past the Deadline or the server closed") + } + } + return nil +} + // algorithm is the syntax of an RFC 8054 algorithm name (5.3): 1 to 20 upper-case letters, // digits, "-" or "_"; case-sensitive. var algorithm = regexp.MustCompile(`^[A-Z0-9_-]{1,20}$`) @@ -139,5 +175,25 @@ func (c *Conn) compressAfter(cmd, status string, outputOnly bool) bool { if cmd != "COMPRESS" || (status != "206" && !strings.HasPrefix(status, "206 ")) || c.Compressed() { return true } - return c.startCompress(outputOnly) == nil + return c.startCompress(outputOnly, c.shape) == nil +} + +// pipeline206 returns the compressed bytes the shape writes in the same socket write as the +// 206 reply to a command (nil for none), and keeps the encoder for startCompress. writeKind +// calls it with the reply's status line; armed is set for the command that starts compression +// (COMPRESS, or any command of a Script step with Compress). +func (c *Conn) pipeline206(armed bool, status string) []byte { + if !armed || !c.shape.pipelined() || c.Compressed() || (status != "206" && !strings.HasPrefix(status, "206 ")) { + return nil + } + enc, err := newEncoder(c.shape) + if err != nil { + return nil + } + b, err := enc.flushData(c.shape.With206) + if err != nil { + return nil + } + c.enc = enc + return slices.Clone(b) } diff --git a/fakeserver/compressshape.go b/fakeserver/compressshape.go new file mode 100644 index 0000000..f95b7e1 --- /dev/null +++ b/fakeserver/compressshape.go @@ -0,0 +1,201 @@ +package fakeserver + +import ( + "bytes" + "compress/flate" + "math/bits" + "slices" + "sort" +) + +// CompressFlush is how the compressed output stream is flushed at the end of every write. +type CompressFlush int + +const ( + // FlushSync ends every write with a sync flush (zlib's Z_SYNC_FLUSH): an empty stored + // block, the marker 00 00 FF FF on a byte boundary. The default. + FlushSync CompressFlush = iota + // FlushPartial ends every write with zlib's Z_PARTIAL_FLUSH, as INN's nnrpd does: an empty + // fixed-Huffman block (10 bits), of which only the whole bytes are sent. The last 0 to 7 + // bits stay in the compressor until the next write; the data before them is complete. + // There is no 00 00 FF FF marker in the stream. The data are literals in fixed-Huffman + // blocks, one per write (no matches), so the output is larger than FlushSync's. A client + // that waits for a stored block, or for the end of the block it is in, hangs on this + // stream, as INN's clients must not. + FlushPartial + // FlushFinish ends the deflate stream (a final block) with the first write after the 206: + // that write is deflated and the final empty block follows in the same socket write. Later + // writes are plain bytes on the socket, after the end of the stream (the kit still inflates + // what the client sends). CompressShape.AfterEnd adds bytes behind the final block. + FlushFinish +) + +// CompressShape shapes what the fake server sends on a COMPRESS DEFLATE connection (RFC 8054), +// for client tests: StoreOptions.CompressShape, PeerOptions.CompressShape, Step.CompressShape, +// Conn.StartCompressShaped. The zero value is the default: every write sync-flushed and sent +// in one socket write. +type CompressShape struct { + // Flush is the flush at the end of each write, see CompressFlush. + Flush CompressFlush + + // Split cuts the compressed bytes of each write (reply, Faults piece) into separate socket + // writes: At holds offsets from the start of the write, a negative offset counts from its + // end (with FlushSync -1 .. -3 cut inside the sync marker 00 00 FF FF, -4 before it, and + // with FlushPartial -1 cuts the empty block's last byte), Size cuts a piece every Size + // bytes when At is empty, Gap pauses after each piece but the last, bounded by the + // server's Deadline like Faults.Split. A client that reads between the pieces sees + // compressed bytes that end mid-block or mid-marker. + Split Split + + // EmptyFlushWith206 writes an empty flush block (the marker of Flush) right after the CRLF + // of the "206" in the same socket write as the 206 line, as a server that flushes its + // compressor when it switches it on. A client must keep those bytes for its inflater. The + // 206 is written by the kit's handlers (Store, Peer, Script step); a Handler that writes + // its own 206 cannot use this. + EmptyFlushWith206 bool + // With206 is deflated and flushed like a reply and written in the same socket write as the + // 206 (it ends with its own flush, so EmptyFlushWith206 adds nothing then): compressed + // payload pipelined behind the 206, for a client that reads past the 206 line. + With206 []byte + + // AfterEnd is written in the same socket write as the final block, with FlushFinish: + // trailing bytes after the end of the stream (junk, or plain text). + AfterEnd []byte +} + +func (s CompressShape) pipelined() bool { return s.EmptyFlushWith206 || len(s.With206) > 0 } + +// cuts returns the sorted offsets in (0, n) where a write of n bytes is cut. +func (s Split) cuts(n int) []int { + var out []int + for _, a := range s.At { + if a < 0 { + a += n + } + if a > 0 && a < n { + out = append(out, a) + } + } + sort.Ints(out) + out = slices.Compact(out) + if len(s.At) == 0 && s.Size > 0 { + for k := s.Size; k < n; k += s.Size { + out = append(out, k) + } + } + return out +} + +// encoder deflates the output of a compressed connection as its shape says. Not safe for +// concurrent use (one connection goroutine writes). +type encoder struct { + shape CompressShape + zw *flate.Writer // FlushSync and FlushFinish; writes into out + out bytes.Buffer + bw bitWriter // FlushPartial + ended bool // FlushFinish: the stream is over +} + +func newEncoder(s CompressShape) (*encoder, error) { + e := &encoder{shape: s} + if s.Flush != FlushPartial { + zw, err := flate.NewWriter(&e.out, flate.DefaultCompression) + if err != nil { + return nil, err + } + e.zw = zw + } + return e, nil +} + +// flushData deflates p and flushes (sync, or partial for FlushPartial; FlushFinish counts as +// sync here). An empty p is an empty flush block. The result is valid until the next call. +func (e *encoder) flushData(p []byte) ([]byte, error) { + if e.shape.Flush == FlushPartial { + return e.partial(p), nil + } + e.out.Reset() + if _, err := e.zw.Write(p); err != nil { // into e.out: cannot fail + return nil, err + } + if err := e.zw.Flush(); err != nil { + return nil, err + } + return e.out.Bytes(), nil +} + +// write returns the bytes to send for a write of p: deflated and flushed, or with FlushFinish +// the end of the stream (once; then p itself). +func (e *encoder) write(p []byte) ([]byte, error) { + if e.ended { + return p, nil + } + if e.shape.Flush != FlushFinish { + return e.flushData(p) + } + e.out.Reset() + if _, err := e.zw.Write(p); err != nil { + return nil, err + } + if err := e.zw.Close(); err != nil { + return nil, err + } + e.ended = true + return append(e.out.Bytes(), e.shape.AfterEnd...), nil +} + +// partial is zlib's Z_PARTIAL_FLUSH (deflate.c: FLUSH_BLOCK, then _tr_align): the data as one +// fixed-Huffman block of literals when there are any, then an empty fixed-Huffman block (3 +// header bits, the 7-bit end-of-block code), and only the whole bytes are sent. Since zlib +// 1.2.5.2 that is all: older versions added a second empty block when fewer than 9 bits of +// the last block's end code had been sent. +func (e *encoder) partial(p []byte) []byte { + w := &e.bw + if len(p) > 0 { + w.block(p) + } + w.block(nil) + return w.take() +} + +// bitWriter writes DEFLATE bits, the first one in the lowest bit of a byte (RFC 1951 3.1.1). +type bitWriter struct { + out []byte + acc uint32 + n uint // bits in acc, fewer than 8 after put +} + +func (w *bitWriter) put(v uint32, n uint) { + w.acc |= v << w.n + w.n += n + for w.n >= 8 { + w.out = append(w.out, byte(w.acc)) + w.acc >>= 8 + w.n -= 8 + } +} + +// code writes an n-bit Huffman code, which goes most significant bit first. +func (w *bitWriter) code(v uint32, n uint) { w.put(bits.Reverse32(v)>>(32-n), n) } + +// block writes a non-final fixed-Huffman block (BFINAL 0, BTYPE 01) of the literals p and the +// end-of-block code (RFC 1951 3.2.6: literals 0-143 are 8 bits from 00110000, 144-255 are 9 +// bits from 110010000, the end of block is 7 zero bits). +func (w *bitWriter) block(p []byte) { + w.put(2, 3) // BFINAL 0 in the lowest bit, BTYPE 01 above it + for _, b := range p { + if b < 144 { + w.code(0x30+uint32(b), 8) + } else { + w.code(0x190+uint32(b)-144, 9) + } + } + w.code(0, 7) +} + +// take returns the whole bytes written so far; the last bits stay. +func (w *bitWriter) take() []byte { + b := w.out + w.out = nil + return b +} diff --git a/fakeserver/compressshape_test.go b/fakeserver/compressshape_test.go new file mode 100644 index 0000000..40dd435 --- /dev/null +++ b/fakeserver/compressshape_test.go @@ -0,0 +1,435 @@ +package fakeserver + +import ( + "bufio" + "bytes" + "compress/flate" + "io" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/go-while/nntp-testkit/internal/inflate" +) + +// Tests of CompressShape: the compressed output stream as the client's raw socket sees it. + +const ( + okLine = "206 Compression active\r\n" + marker = "\x00\x00\xff\xff" // the sync flush marker +) + +// recConn records every write to the socket (the pieces of the 206 line and of the compressed +// stream), so a test can tell what was one write without relying on read boundaries. +type recConn struct { + net.Conn + mu sync.Mutex + w [][]byte +} + +func (r *recConn) Write(p []byte) (int, error) { + n, err := r.Conn.Write(p) + r.mu.Lock() + r.w = append(r.w, bytes.Clone(p[:n])) + r.mu.Unlock() + return n, err +} + +func (r *recConn) writes() [][]byte { + r.mu.Lock() + defer r.mu.Unlock() + return append([][]byte(nil), r.w...) +} + +// shapeServer is a Store with Compress and shape; DATE answers dateReply. The recConn sees +// the writes from the greeting on that follow the handler's start, so w[0] is the 206 write. +func shapeServer(t *testing.T, shape CompressShape) (*Server, *recConn) { + t.Helper() + rec := &recConn{} + h := StoreHandler(testStore(), StoreOptions{Compress: true, CompressShape: shape}) + s := New(t, Options{Deadline: 10 * time.Second, + Faults: Faults{Override: map[string][]string{"DATE": {"111 12345678901234"}}}, + Handler: func(c *Conn) { + rec.Conn = c.conn + c.conn = rec + h(c) + }}) + return s, rec +} + +// zClient is a raw client of a compressed connection: its commands are deflated and sync +// flushed in one write. +type zClient struct { + t *testing.T + c net.Conn + br *bufio.Reader // reads c behind the 206 line; use it once the test reads the compressed stream + zw *flate.Writer + b bytes.Buffer +} + +// zDial connects, sends COMPRESS DEFLATE and reads the 206 line byte by byte, so that bytes +// behind it stay in the socket. +func zDial(t *testing.T, s *Server) *zClient { + t.Helper() + c := bytesDial(t, s) + rawSend(t, c, "COMPRESS DEFLATE\r\n") + var line []byte + one := make([]byte, 1) + for !bytes.HasSuffix(line, []byte("\r\n")) { + if _, err := io.ReadFull(c, one); err != nil { + t.Fatalf("reading the 206: %v (got %q)", err, line) + } + line = append(line, one[0]) + } + if string(line) != okLine { + t.Fatalf("COMPRESS: %q", line) + } + z := &zClient{t: t, c: c, br: bufio.NewReader(c)} + z.zw, _ = flate.NewWriter(&z.b, flate.DefaultCompression) + return z +} + +func (z *zClient) send(cmd string) { + z.t.Helper() + z.b.Reset() + _, _ = z.zw.Write([]byte(cmd + "\r\n")) + _ = z.zw.Flush() + if _, err := z.c.Write(z.b.Bytes()); err != nil { + z.t.Fatal(err) + } +} + +// refStream deflates replies the way the default shape does (one default-level flate writer, +// sync flush after each reply), to know the bytes a test expects on the wire. +type refStream struct { + b bytes.Buffer + zw *flate.Writer +} + +func newRef() *refStream { + r := &refStream{} + r.zw, _ = flate.NewWriter(&r.b, flate.DefaultCompression) + return r +} + +func (r *refStream) flush(p string) []byte { + r.b.Reset() + _, _ = r.zw.Write([]byte(p)) + _ = r.zw.Flush() + return bytes.Clone(r.b.Bytes()) +} + +// plain reads the compressed stream of c with the kit's inflater, which passes on the bytes as +// soon as they can be decoded (so it also reads partial-flush streams), and returns a reader +// of the plaintext. +func plain(t *testing.T, z *zClient) *bufio.Reader { + t.Helper() + pr, pw := io.Pipe() + t.Cleanup(func() { _ = pr.Close() }) + go func() { + _ = pw.CloseWithError(inflate.Run(z.br, func(b []byte) error { + _, err := pw.Write(b) + return err + })) + }() + return bufio.NewReader(pr) +} + +func wantPlainLine(t *testing.T, br *bufio.Reader, c net.Conn, want string) { + t.Helper() + _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) + line, err := br.ReadString('\n') + if err != nil || line != want { + t.Fatalf("reply %q, %v; want %q", line, err, want) + } +} + +// The default shape: one socket write per reply, a sync flush at its end. +func TestCompressShapeDefault(t *testing.T) { + s, rec := shapeServer(t, CompressShape{}) + z := zDial(t, s) + br := plain(t, z) + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + w := rec.writes() + ref := newRef() + if len(w) != 3 || string(w[0]) != okLine || !bytes.Equal(w[1], ref.flush(dateReply)) || !bytes.Equal(w[2], ref.flush(dateReply)) { + t.Fatalf("writes %q", w) + } + if !bytes.HasSuffix(w[1], []byte(marker)) { + t.Errorf("no sync marker at the end of %x", w[1]) + } +} + +func TestCompressShapeSplitPieces(t *testing.T) { + const gap = 120 * time.Millisecond + ref := newRef() + first, second := ref.flush(dateReply), ref.flush(dateReply) + n := len(first) + for _, tc := range []struct { + name string + split Split + cuts []int + }{ + {"offsets from start and end", Split{At: []int{2, -3}, Gap: gap}, []int{2, n - 3}}, + {"inside the marker", Split{At: []int{-1, -2, -3}, Gap: gap}, []int{n - 3, n - 2, n - 1}}, + {"size", Split{Size: 7, Gap: gap}, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + cuts := tc.cuts + if cuts == nil { + for k := tc.split.Size; k < n; k += tc.split.Size { + cuts = append(cuts, k) + } + } + var pieces []string + prev := 0 + for _, k := range append(append([]int{}, cuts...), n) { + pieces = append(pieces, string(first[prev:k])) + prev = k + } + s, rec := shapeServer(t, CompressShape{Split: tc.split}) + z := zDial(t, s) + start := time.Now() + z.send("DATE") + readPieces(t, z.c, start, pieces, gaps(len(pieces), gap)) + // the pieces are exactly the socket writes of the server + w := rec.writes() + if len(w) != 1+len(pieces) { + t.Fatalf("%d writes, want the 206 and %d pieces", len(w), len(pieces)) + } + for i, p := range pieces { + if string(w[1+i]) != p { + t.Errorf("write %d = %x, want %x", i, w[1+i], p) + } + } + // the stream goes on: the next reply is deflated after the first as the reference does + start = time.Now() + z.send("DATE") + readPieces(t, z.c, start, []string{string(second)}, []time.Duration{0}) + }) + } +} + +// A client inflater (compress/flate's reader) must cope with the stream arriving byte by byte. +func TestCompressShapeSplitBytewise(t *testing.T) { + s, _ := shapeServer(t, CompressShape{Split: Split{Size: 1}}) + z := zDial(t, s) + zr := flate.NewReader(z.br) + for range 3 { + z.send("DATE") + buf := make([]byte, len(dateReply)) + _ = z.c.SetReadDeadline(time.Now().Add(10 * time.Second)) + if _, err := io.ReadFull(zr, buf); err != nil || string(buf) != dateReply { + t.Fatalf("reply %q, %v", buf, err) + } + } +} + +func TestCompressShapeEmptyFlushWith206(t *testing.T) { + for _, tc := range []struct { + name string + shape CompressShape + empty string + }{ + {"sync", CompressShape{EmptyFlushWith206: true}, "\x00" + marker}, + {"partial", CompressShape{Flush: FlushPartial, EmptyFlushWith206: true}, "\x02"}, + } { + t.Run(tc.name, func(t *testing.T) { + s, rec := shapeServer(t, tc.shape) + z := zDial(t, s) + // the bytes behind the 206 line are already here: nothing was sent after the line + _ = z.c.SetReadDeadline(time.Now().Add(5 * time.Second)) + if got, err := z.br.Peek(len(tc.empty)); err != nil || string(got) != tc.empty { + t.Fatalf("bytes after the 206: %x, %v; want %x", got, err, tc.empty) + } + if w := rec.writes(); len(w) != 1 || string(w[0]) != okLine+tc.empty { + t.Fatalf("the 206 and the empty flush are not one write: %q", w) + } + // the empty block is part of the stream: replies follow it + br := plain(t, z) + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + }) + } +} + +// Without the option nothing follows the 206 line. +func TestCompressShapeNoEmptyFlush(t *testing.T) { + s, rec := shapeServer(t, CompressShape{}) + z := zDial(t, s) + if !readQuiet(t, z.c, 200*time.Millisecond) { + t.Fatal("bytes after the 206") + } + if w := rec.writes(); len(w) != 1 || string(w[0]) != okLine { + t.Fatalf("writes %q", w) + } +} + +// With206 pipelines a deflated payload in the write of the 206 line (the client reads it +// before it has sent a command). +func TestCompressShapeWith206(t *testing.T) { + s, rec := shapeServer(t, CompressShape{With206: []byte("hello\r\n")}) + z := zDial(t, s) + br := plain(t, z) + wantPlainLine(t, br, z.c, "hello\r\n") + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + w := rec.writes() + if string(w[0][:len(okLine)]) != okLine || len(w[0]) == len(okLine) { + t.Fatalf("first write %q", w[0]) + } +} + +// Partial flush as zlib's: replies decode, promptly, and the stream has no sync marker. +func TestCompressShapePartialFlush(t *testing.T) { + s, rec := shapeServer(t, CompressShape{Flush: FlushPartial}) + z := zDial(t, s) + br := plain(t, z) + for range 3 { + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) // arrives although the last bits of the flush block are not sent + } + z.send("HELP") + wantPlainLine(t, br, z.c, "100 Help text follows\r\n") + var stream []byte + for _, p := range rec.writes()[1:] { + stream = append(stream, p...) + } + if bytes.Contains(stream, []byte(marker)) { + t.Errorf("sync marker in a partial-flush stream: %x", stream) + } +} + +// The partial-flush encoder against compress/flate's reader, which has no partial flush of its +// own but decodes the stream once its last block is completed. +func TestPartialEncoderDecodes(t *testing.T) { + e, err := newEncoder(CompressShape{Flush: FlushPartial}) + if err != nil { + t.Fatal(err) + } + var plainText, stream []byte + all := make([]byte, 256) // every byte value: the 8-bit and the 9-bit codes + for i := range all { + all[i] = byte(i) + } + for _, p := range [][]byte{[]byte("211 6 1 6 compress.test\r\n"), nil, all, []byte("."), {0xff, 0xff, 0x00}, bytes.Repeat([]byte("xyz\xf0"), 5000)} { + b, _ := e.flushData(p) + stream = append(stream, b...) + plainText = append(plainText, p...) + } + if bytes.Contains(stream, []byte(marker)) { + t.Error("sync marker in the stream") + } + // complete the stream: pad the unsent bits (zeros, the end of the empty block's code) and + // add a final empty stored block + if e.bw.n > 0 { + e.bw.put(0, 8-e.bw.n) + } + stream = append(append(stream, e.bw.take()...), 0x01, 0x00, 0x00, 0xff, 0xff) + got, err := io.ReadAll(flate.NewReader(bytes.NewReader(stream))) + if err != nil || !bytes.Equal(got, plainText) { + t.Fatalf("decoded %d bytes, %v; want %d", len(got), err, len(plainText)) + } + // the kit's inflater decodes the unfinished stream as it is + var out []byte + _ = inflate.Run(bufio.NewReader(bytes.NewReader(stream[:len(stream)-5])), func(b []byte) error { out = append(out, b...); return nil }) + if !bytes.Equal(out, plainText) { + t.Errorf("kit inflater: %d bytes, want %d", len(out), len(plainText)) + } +} + +// The first sync flush of an empty stream and the first partial flush have the bytes the +// tests above expect (a change of compress/flate would show here first). +func TestEmptyFlushBytes(t *testing.T) { + e, _ := newEncoder(CompressShape{}) + if b, _ := e.flushData(nil); string(b) != "\x00"+marker { + t.Errorf("sync empty flush %x", b) + } + e, _ = newEncoder(CompressShape{Flush: FlushPartial}) + if b, _ := e.flushData(nil); string(b) != "\x02" { + t.Errorf("partial empty flush %x", b) + } +} + +func TestCompressShapeFinish(t *testing.T) { + for _, after := range []string{"", "junk\r\nmore"} { + t.Run("after="+after, func(t *testing.T) { + s, rec := shapeServer(t, CompressShape{Flush: FlushFinish, AfterEnd: []byte(after)}) + z := zDial(t, s) + z.send("DATE") + br := z.br // a ByteReader: flate reads no byte past the final block + _ = z.c.SetReadDeadline(time.Now().Add(10 * time.Second)) + got, err := io.ReadAll(flate.NewReader(br)) + if err != nil || string(got) != dateReply { + t.Fatalf("stream %q, %v", got, err) + } + rest := make([]byte, len(after)) + if _, err := io.ReadFull(br, rest); err != nil || string(rest) != after { + t.Fatalf("after the end %q, %v", rest, err) + } + if w := rec.writes(); len(w) != 2 || !bytes.HasSuffix(w[1], []byte(after)) { + t.Fatalf("the final block and the trailing bytes are not one write: %q", w) + } + // the client's input is still inflated, the server's output is plain from here on + z.send("DATE") + line, err := br.ReadString('\n') + if err != nil || line != dateReply { + t.Fatalf("plain reply %q, %v", line, err) + } + }) + } +} + +// The shape also applies to a Peer and to a Script step. +func TestCompressShapePeerAndScript(t *testing.T) { + shape := CompressShape{Flush: FlushPartial, EmptyFlushWith206: true, Split: Split{Size: 2}} + t.Run("peer", func(t *testing.T) { + s := New(t, Options{Handler: PeerHandler(&PeerOptions{Compress: true, CompressShape: shape})}) + z := zDial(t, s) + if _, err := z.br.Peek(1); err != nil { // the empty block + t.Fatal(err) + } + br := plain(t, z) + z.send("MODE STREAM") + line, err := br.ReadString('\n') + if err != nil || !strings.HasPrefix(line, "203 ") { + t.Fatalf("%q, %v", line, err) + } + }) + t.Run("script", func(t *testing.T) { + s := New(t, Options{Script: []Step{ + {Expect: "COMPRESS DEFLATE", Reply: []string{"206 Compression active"}, Compress: true, CompressShape: shape}, + {Expect: "DATE", Reply: []string{"111 20260102030405"}}, + }}) + z := zDial(t, s) + if _, err := z.br.Peek(1); err != nil { + t.Fatal(err) + } + br := plain(t, z) + z.send("DATE") + wantPlainLine(t, br, z.c, "111 20260102030405\r\n") + }) +} + +// StartCompressShaped from a Handler that writes its own 206. +func TestStartCompressShaped(t *testing.T) { + s := New(t, Options{Handler: func(c *Conn) { + if _, err := c.ReadLine(); err != nil { + return + } + _ = c.WriteLines("206 Compression active") + if c.StartCompressShaped(CompressShape{Flush: FlushPartial}) != nil || c.StartCompressShaped(CompressShape{}) == nil { + return + } + _ = c.WriteLines("111 20260102030405") + c.drain() + }}) + z := zDial(t, s) + br := plain(t, z) + wantPlainLine(t, br, z.c, "111 20260102030405\r\n") +} diff --git a/fakeserver/conn.go b/fakeserver/conn.go index d99de45..2a1d78d 100644 --- a/fakeserver/conn.go +++ b/fakeserver/conn.go @@ -36,6 +36,10 @@ type Conn struct { cmd string // command of the last client line read with ReadLine (the key of the byte-level faults) faulted bool // a fault closes the connection now (Faults.Abort decides how) + + shape CompressShape // the shape of the stream compressAfter starts (the handler's CompressShape) + shapeStep bool // Script: the step being served starts compression, whatever its command + enc *encoder // made when the 206 carried pipelined bytes; startCompress takes it over } // ID returns the connection's number in accept order, starting at 1 (connections rejected by @@ -318,6 +322,7 @@ func (c *Conn) writeKind(line string, kind replyKind, lines []string) bool { if len(lines) > 0 { c.last = lines[0] } + tail := c.pipeline206(cmd == "COMPRESS" || c.shapeStep, c.last) // empty flush etc. in the 206's write endless := normal && f.Endless[cmd] if endless && len(lines) > 0 && lines[len(lines)-1] == "." { lines = lines[:len(lines)-1] @@ -343,11 +348,15 @@ func (c *Conn) writeKind(line string, kind replyKind, lines []string) bool { if i > 0 && (time.Now().After(end) || !c.sleep(pause)) { return c.cutIf(w) } - if !w.write(joinCRLF([]string{l})) { + b := joinCRLF([]string{l}) + if i == len(lines)-1 { + b = append(b, tail...) + } + if !w.write(b) { return c.cutIf(w) } } - } else if len(lines) > 0 && !w.write(joinCRLF(lines)) { + } else if len(lines) > 0 && !w.write(append(joinCRLF(lines), tail...)) { return c.cutIf(w) } if endless { diff --git a/fakeserver/doc.go b/fakeserver/doc.go index c26d5f6..8921654 100644 --- a/fakeserver/doc.go +++ b/fakeserver/doc.go @@ -25,11 +25,12 @@ // // NoGreeting; then Handler when set, else Store when set, else Script (with NoGreeting and // // no Handler the Script/Store is not run: lines are only recorded). // -// type Step struct { Expect string; ReadDot bool; Delay time.Duration; Reply []string; Hang, CloseAfter, Compress bool } +// type Step struct { Expect string; ReadDot bool; Delay time.Duration; Reply []string; Hang, CloseAfter, Compress bool; CompressShape CompressShape } // // Semantics as go-pugleaf's cliFakeStep: Expect is an anchored regexp ("" = any line); a // // non-matching line gets "500 fakeserver: unexpected " and is recorded in Mismatches(); // // after the last step the connection keeps reading and recording until EOF. Compress -// // switches the connection to COMPRESS DEFLATE after the step's reply (Conn.StartCompress). +// // switches the connection to COMPRESS DEFLATE after the step's reply (Conn.StartCompress), +// // CompressShape shapes its output stream. // // type Server struct { ... } // func New(tb testing.TB, o Options) *Server // listens, serves, tb.Cleanup(Close); fails tb on listen error @@ -59,6 +60,7 @@ // func (c *Conn) ReadDotBlockRaw() ([]string, []byte, error) // the same, also the exact bytes of the block (not the "." line) // func (c *Conn) Raw() net.Conn // the TLS conn when TLS, else the TCP conn; after StartCompress the compressing conn on top // func (c *Conn) StartCompress() error // RFC 8054 COMPRESS DEFLATE from now on, both directions; call it after writing the 206; error when already on +// func (c *Conn) StartCompressShaped(s CompressShape) error // StartCompress with the output stream shaped, see CompressShape below // func (c *Conn) StartCompressOutputOnly() error // broken COMPRESS: deflate only the server's output, read the client's bytes uncompressed // func (c *Conn) Compressed() bool // func (c *Conn) Server() *Server @@ -93,6 +95,7 @@ // OnArticleRaw func(msgID string, raw []byte) // right after OnArticle: the article's exact stuffed bytes, no terminator (article.ParseDotBlock) // Compress bool // offer COMPRESS DEFLATE as StoreOptions.Compress; AUTHINFO answers 502 once active // CompressOutputOnly bool // offer it, but after the 206 compress only the output (Conn.StartCompressOutputOnly) +// CompressShape CompressShape // shape the compressed output stream, see below // HoldGreeting <-chan struct{} // greeting waits for close/receive; needs Options.NoGreeting (NewPeer sets it); the hold ends at Deadline (60 s default) and then the connection is dropped without a greeting, as on Close; the conn counts as open // GreetingDelay time.Duration // sleep before the greeting (after HoldGreeting), capped at Deadline; needs Options.NoGreeting (NewPeer sets it) // BeforeReply func(c *Conn, cmd, line, status string) // before every reply (IHAVE: 335 and the final one); may block, e.g. c.Wait(ch) @@ -167,6 +170,7 @@ // OverviewFields func(conn int, group string, fields []string) []string // rewrite the fields of every OVER line (overview out of step with the articles) // Compress bool // offer COMPRESS DEFLATE (RFC 8054), see below // CompressOutputOnly bool // offer it broken: after the 206 only the output is compressed, the client's input is read uncompressed +// CompressShape CompressShape // shape the compressed output stream, see below // CompressBeforeAuth bool // with User/Pass: list and take COMPRESS before AUTHINFO (INN's nnrpd); other commands still 480, AUTHINFO 502 once compressed // StartTLS *StartTLSOptions // offer STARTTLS, see below // } @@ -269,4 +273,19 @@ // client tests: the 206 and everything after it is compressed, but the client's compressed // commands are read as they are, so a client that follows the RFC gets no answer or "500" // answers to binary lines. +// +// CompressShape (StoreOptions, PeerOptions, Step, Conn.StartCompressShaped) shapes the output +// stream for client tests; the zero value is the behaviour above: +// +// type CompressShape struct { +// Flush CompressFlush // FlushSync (default); FlushPartial: zlib's Z_PARTIAL_FLUSH like INN's nnrpd (an empty fixed-Huffman block, no 00 00 FF FF marker, the last 0-7 bits unsent); FlushFinish: the first write after the 206 ends the stream, later writes are plain +// Split Split // cut the compressed bytes of each write into socket writes: At (negative = from the end, -1..-3 inside the sync marker), Size, Gap (bounded by Deadline) +// EmptyFlushWith206 bool // the 206 line and an empty flush block in one socket write (Store, Peer, Script; not a Handler's own 206) +// With206 []byte // deflated and flushed into the 206's socket write (compressed payload pipelined behind the 206) +// AfterEnd []byte // with FlushFinish: bytes after the final block, in its socket write +// } +// +// FlushPartial writes literals in fixed-Huffman blocks (no matches) with a small bit writer, +// since compress/flate cannot do a partial flush; compress/flate's reader cannot read such a +// stream before the end either (it holds data until a stored block), the kit's inflater can. package fakeserver diff --git a/fakeserver/peer.go b/fakeserver/peer.go index 7c68a16..c53517f 100644 --- a/fakeserver/peer.go +++ b/fakeserver/peer.go @@ -30,6 +30,9 @@ type PeerOptions struct { // peer's output and reads the client's input uncompressed (Conn.StartCompressOutputOnly): // a broken peer, for client tests. CompressOutputOnly bool + // CompressShape shapes the compressed output stream after the 206 (flush kind, split + // writes, an empty flush with the 206, a final block); the zero value is the default. + CompressShape CompressShape // Reject are the articles the peer refuses once it has read them, as a server whose // filter refuses the content does: CHECK still answers 238 (the id alone is wanted), @@ -130,6 +133,7 @@ func PeerHandler(o *PeerOptions) func(*Conn) { dropReply = "400 Service temporarily unavailable" } return func(c *Conn) { + c.shape = o.CompressShape if !o.greet(c) { return } diff --git a/fakeserver/server.go b/fakeserver/server.go index c8a1574..913d2f9 100644 --- a/fakeserver/server.go +++ b/fakeserver/server.go @@ -68,6 +68,9 @@ type Step struct { Hang bool // after the match (and ReadDot, Delay, Reply), never reply again; held until Close CloseAfter bool // close the connection after the reply Compress bool // after the reply (e.g. "206 Compression active"), compress both directions (Conn.StartCompress) + // CompressShape shapes the compressed output stream of a Compress step (see CompressShape); + // EmptyFlushWith206 and With206 go with the step's reply when that is a 206. + CompressShape CompressShape } // StoreOptions configures the store handler (Options.StoreOpts, StoreHandler). @@ -137,6 +140,13 @@ type StoreOptions struct { // inflating it (Conn.StartCompressOutputOnly): a broken server, for client tests. CompressOutputOnly bool + // CompressShape shapes the compressed output stream after the 206: the flush (sync, zlib's + // partial flush like INN's nnrpd, or a final block with trailing bytes), the compressed + // bytes cut into several socket writes, an empty flush in the packet of the 206 line. The + // zero value is the default: every write sync-flushed and sent in one socket write. It + // applies to Compress and CompressOutputOnly (see CompressShape). + CompressShape CompressShape + // CompressBeforeAuth makes a server with User or Pass set list "COMPRESS DEFLATE" before // authentication and take COMPRESS then, as INN's nnrpd does (COMPRESS is not a reading // command): the other commands but CAPABILITIES, MODE, HELP, AUTHINFO, DATE and QUIT still @@ -647,10 +657,13 @@ func (s *Server) runScript(c *Conn) { if st.Delay > 0 && !c.sleep(st.Delay) { return } - if !c.writeReply(line, st.Reply...) { + c.shape, c.shapeStep = st.CompressShape, st.Compress + ok := c.writeReply(line, st.Reply...) + c.shapeStep = false + if !ok { return } - if st.Compress && !c.Compressed() && c.StartCompress() != nil { + if st.Compress && !c.Compressed() && c.startCompress(false, st.CompressShape) != nil { return } if st.Hang { diff --git a/fakeserver/storehandler.go b/fakeserver/storehandler.go index 59fc3ff..6cc8c5f 100644 --- a/fakeserver/storehandler.go +++ b/fakeserver/storehandler.go @@ -91,6 +91,7 @@ func StoreHandler(st *Store, o StoreOptions) func(*Conn) { seq := new(atomic.Int64) // generated POST Message-IDs, shared by the handler's connections postMu := new(sync.Mutex) // serialises filing POSTs across the handler's connections return func(c *Conn) { + c.shape = o.CompressShape h := &storeConn{st: st, o: o, c: c, seq: seq, postMu: postMu, authed: o.User == "" && o.Pass == ""} h.serve() } From 27d406c45398451fb999aa60ec360881762b4758 Mon Sep 17 00:00:00 2001 From: go-while Date: Thu, 12 Nov 2026 13:56:05 +0000 Subject: [PATCH 2/2] fakeserver: review fixes for CompressShape Arm the 206 tail only when compression will start, keep the caller's write deadline across a split gap, record writes before they happen in the test recorder, terminate the partial-flush test stream correctly, document the unsent bits and the tail in fault offsets. Co-Authored-By: Claude Opus 5.5 --- fakeserver/compress.go | 19 +++++-- fakeserver/compressshape.go | 8 +-- fakeserver/compressshape_test.go | 87 +++++++++++++++++++++++++++++--- fakeserver/conn.go | 3 +- fakeserver/peer.go | 2 +- fakeserver/storehandler.go | 2 +- 6 files changed, 106 insertions(+), 15 deletions(-) diff --git a/fakeserver/compress.go b/fakeserver/compress.go index 7994231..52d5a83 100644 --- a/fakeserver/compress.go +++ b/fakeserver/compress.go @@ -87,7 +87,20 @@ type deflateConn struct { enc *encoder sleep func(time.Duration) bool // Conn.sleep: bounded by the Deadline, false when the server closed deadline time.Duration - werr error // the first failed write: the stream is broken from there on + werr error // the first failed write: the stream is broken from there on + wdl time.Time // the write deadline last set through this conn (zero: none) +} + +// SetWriteDeadline and SetDeadline remember the write deadline, so that send can put back the +// one writeKind set (Drip, LineDelay, Split, Endless) instead of clearing it. +func (d *deflateConn) SetWriteDeadline(t time.Time) error { + d.wdl = t + return d.Conn.SetWriteDeadline(t) +} + +func (d *deflateConn) SetDeadline(t time.Time) error { + d.wdl = t + return d.Conn.SetDeadline(t) } func (d *deflateConn) Read(p []byte) (int, error) { return d.r.Read(p) } @@ -133,9 +146,9 @@ func (d *deflateConn) send(b []byte) error { return err } end := time.Now().Add(d.deadline) - if sp.Gap > 0 { + if sp.Gap > 0 && (d.wdl.IsZero() || end.Before(d.wdl)) { _ = d.Conn.SetWriteDeadline(end) - defer d.Conn.SetWriteDeadline(time.Time{}) + defer d.Conn.SetWriteDeadline(d.wdl) // what the caller had set, not "none" } prev := 0 for _, k := range append(cuts, len(b)) { diff --git a/fakeserver/compressshape.go b/fakeserver/compressshape.go index f95b7e1..b05a00b 100644 --- a/fakeserver/compressshape.go +++ b/fakeserver/compressshape.go @@ -18,7 +18,8 @@ const ( // FlushPartial ends every write with zlib's Z_PARTIAL_FLUSH, as INN's nnrpd does: an empty // fixed-Huffman block (10 bits), of which only the whole bytes are sent. The last 0 to 7 // bits stay in the compressor until the next write; the data before them is complete. - // There is no 00 00 FF FF marker in the stream. The data are literals in fixed-Huffman + // The data before them is complete: the held bits are only the tail of the empty block's + // end-of-block code, which is zeros. There is no 00 00 FF FF marker in the stream. The data are literals in fixed-Huffman // blocks, one per write (no matches), so the output is larger than FlushSync's. A client // that waits for a stored block, or for the end of the block it is in, hangs on this // stream, as INN's clients must not. @@ -50,8 +51,9 @@ type CompressShape struct { // EmptyFlushWith206 writes an empty flush block (the marker of Flush) right after the CRLF // of the "206" in the same socket write as the 206 line, as a server that flushes its // compressor when it switches it on. A client must keep those bytes for its inflater. The - // 206 is written by the kit's handlers (Store, Peer, Script step); a Handler that writes - // its own 206 cannot use this. + // 206 is written by the kit's handlers (Store, Peer with Compress or CompressOutputOnly, + // Script step with Compress); a Handler that writes its own 206 cannot use this. Faults on + // the COMPRESS reply (DropAfterBytes, Split offsets) count these bytes too. EmptyFlushWith206 bool // With206 is deflated and flushed like a reply and written in the same socket write as the // 206 (it ends with its own flush, so EmptyFlushWith206 adds nothing then): compressed diff --git a/fakeserver/compressshape_test.go b/fakeserver/compressshape_test.go index 40dd435..3606361 100644 --- a/fakeserver/compressshape_test.go +++ b/fakeserver/compressshape_test.go @@ -30,10 +30,17 @@ type recConn struct { } func (r *recConn) Write(p []byte) (int, error) { - n, err := r.Conn.Write(p) + // recorded before the write, so a client that has read the bytes finds them in writes() r.mu.Lock() - r.w = append(r.w, bytes.Clone(p[:n])) + r.w = append(r.w, bytes.Clone(p)) + i := len(r.w) - 1 r.mu.Unlock() + n, err := r.Conn.Write(p) + if n < len(p) { + r.mu.Lock() + r.w[i] = r.w[i][:n] + r.mu.Unlock() + } return n, err } @@ -325,19 +332,22 @@ func TestPartialEncoderDecodes(t *testing.T) { if bytes.Contains(stream, []byte(marker)) { t.Error("sync marker in the stream") } - // complete the stream: pad the unsent bits (zeros, the end of the empty block's code) and - // add a final empty stored block + // complete the stream: the held bits are the zero tail of the empty block's end code; a + // final stored block header (BFINAL 1, BTYPE 00) follows them at once, then padding to a + // byte and the empty block's LEN and NLEN + unfinished := len(stream) + e.bw.put(1, 3) if e.bw.n > 0 { e.bw.put(0, 8-e.bw.n) } - stream = append(append(stream, e.bw.take()...), 0x01, 0x00, 0x00, 0xff, 0xff) + stream = append(append(stream, e.bw.take()...), 0x00, 0x00, 0xff, 0xff) got, err := io.ReadAll(flate.NewReader(bytes.NewReader(stream))) if err != nil || !bytes.Equal(got, plainText) { t.Fatalf("decoded %d bytes, %v; want %d", len(got), err, len(plainText)) } // the kit's inflater decodes the unfinished stream as it is var out []byte - _ = inflate.Run(bufio.NewReader(bytes.NewReader(stream[:len(stream)-5])), func(b []byte) error { out = append(out, b...); return nil }) + _ = inflate.Run(bufio.NewReader(bytes.NewReader(stream[:unfinished])), func(b []byte) error { out = append(out, b...); return nil }) if !bytes.Equal(out, plainText) { t.Errorf("kit inflater: %d bytes, want %d", len(out), len(plainText)) } @@ -433,3 +443,68 @@ func TestStartCompressShaped(t *testing.T) { br := plain(t, z) wantPlainLine(t, br, z.c, "111 20260102030405\r\n") } + +// The pipelined tail is armed only when compression starts: a Script step without Compress and +// a Peer without Compress (a forced 206) stay plain. +func TestCompressShapeNotArmedWithoutCompress(t *testing.T) { + shape := CompressShape{EmptyFlushWith206: true, With206: []byte("x")} + check := func(t *testing.T, s *Server, cmd, wantPrefix string) { + t.Helper() + c := bytesDial(t, s) + rawSend(t, c, cmd+"\r\n") + br := bufio.NewReader(c) + line, err := br.ReadString('\n') + if err != nil || !strings.HasPrefix(line, "206 ") { + t.Fatalf("%q, %v", line, err) + } + if br.Buffered() != 0 || !readQuiet(t, c, 200*time.Millisecond) { + t.Fatal("bytes after the 206 of a connection that stays plain") + } + rawSend(t, c, "MODE STREAM\r\n") + _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) + if line, err = br.ReadString('\n'); err != nil || !strings.HasPrefix(line, wantPrefix) { + t.Fatalf("plain command after the 206: %q, %v", line, err) + } + } + t.Run("script", func(t *testing.T) { + s := New(t, Options{Script: []Step{ + {Expect: "COMPRESS DEFLATE", Reply: []string{"206 ok"}, CompressShape: shape}, + {Expect: "MODE STREAM", Reply: []string{"203 ok"}}, + }}) + check(t, s, "COMPRESS DEFLATE", "203 ") + }) + t.Run("peer", func(t *testing.T) { + s := New(t, Options{Handler: PeerHandler(&PeerOptions{CompressShape: shape}), + Faults: Faults{Override: map[string][]string{"COMPRESS": {"206 ok"}}}}) + check(t, s, "COMPRESS DEFLATE", "203 ") + }) +} + +// A write deadline that the caller set (writeKind does for its pauses) survives a Write with a +// Split gap; without one, the deadline set for the pauses is cleared again. +type dlConn struct { + net.Conn + dl []time.Time +} + +func (c *dlConn) Write(p []byte) (int, error) { return len(p), nil } +func (c *dlConn) SetWriteDeadline(t time.Time) error { c.dl = append(c.dl, t); return nil } + +func TestCompressSplitKeepsCallerDeadline(t *testing.T) { + for _, set := range []bool{true, false} { + sc := &dlConn{} + enc, _ := newEncoder(CompressShape{Split: Split{Size: 3, Gap: time.Millisecond}}) + d := &deflateConn{Conn: sc, enc: enc, sleep: func(time.Duration) bool { return true }, deadline: time.Hour} + var want time.Time + if set { + want = time.Now().Add(time.Minute) + _ = d.SetWriteDeadline(want) + } + if _, err := d.Write([]byte("hello\r\n")); err != nil { + t.Fatal(err) + } + if last := sc.dl[len(sc.dl)-1]; !last.Equal(want) { + t.Errorf("caller deadline set=%v: after Write the deadline is %v, want %v", set, last, want) + } + } +} diff --git a/fakeserver/conn.go b/fakeserver/conn.go index 2a1d78d..97b3c43 100644 --- a/fakeserver/conn.go +++ b/fakeserver/conn.go @@ -38,6 +38,7 @@ type Conn struct { faulted bool // a fault closes the connection now (Faults.Abort decides how) shape CompressShape // the shape of the stream compressAfter starts (the handler's CompressShape) + shapeOn bool // Store/Peer: COMPRESS is offered, so a 206 to it starts compression shapeStep bool // Script: the step being served starts compression, whatever its command enc *encoder // made when the 206 carried pipelined bytes; startCompress takes it over } @@ -322,7 +323,7 @@ func (c *Conn) writeKind(line string, kind replyKind, lines []string) bool { if len(lines) > 0 { c.last = lines[0] } - tail := c.pipeline206(cmd == "COMPRESS" || c.shapeStep, c.last) // empty flush etc. in the 206's write + tail := c.pipeline206((cmd == "COMPRESS" && c.shapeOn) || c.shapeStep, c.last) // empty flush etc. in the 206's write endless := normal && f.Endless[cmd] if endless && len(lines) > 0 && lines[len(lines)-1] == "." { lines = lines[:len(lines)-1] diff --git a/fakeserver/peer.go b/fakeserver/peer.go index c53517f..fbe3a40 100644 --- a/fakeserver/peer.go +++ b/fakeserver/peer.go @@ -133,7 +133,7 @@ func PeerHandler(o *PeerOptions) func(*Conn) { dropReply = "400 Service temporarily unavailable" } return func(c *Conn) { - c.shape = o.CompressShape + c.shape, c.shapeOn = o.CompressShape, offerCompress if !o.greet(c) { return } diff --git a/fakeserver/storehandler.go b/fakeserver/storehandler.go index 6cc8c5f..a44e5b8 100644 --- a/fakeserver/storehandler.go +++ b/fakeserver/storehandler.go @@ -91,7 +91,7 @@ func StoreHandler(st *Store, o StoreOptions) func(*Conn) { seq := new(atomic.Int64) // generated POST Message-IDs, shared by the handler's connections postMu := new(sync.Mutex) // serialises filing POSTs across the handler's connections return func(c *Conn) { - c.shape = o.CompressShape + c.shape, c.shapeOn = o.CompressShape, o.Compress || o.CompressOutputOnly h := &storeConn{st: st, o: o, c: c, seq: seq, postMu: postMu, authed: o.User == "" && o.Pass == ""} h.serve() }