THRIFT-6281: Hold the frames TFramedTransport writes to the configured frame size

lib/go/README.md

@@ -287,3 +287,12 @@ larger `MaxFrameSize` in its own configuration, not only in the server's:
     conf := &thrift.TConfiguration{
         MaxFrameSize: 64 * 1024 * 1024,
     }
+
+A note about the frames TFramedTransport writes
+===============================================
+
+`TFramedTransport.Flush` holds the frames it writes to the same limit, through
+the same check. A larger frame is not written: `Flush` returns a
+`TProtocolException` of type `SIZE_LIMIT` and drops the frame, where it
+previously returned no error and wrote a frame `readFrame` refuses. The note
+above on raising `MaxFrameSize` applies to it as well.

lib/go/thrift/framed_transport.go

@@ -202,11 +202,15 @@ func (p *TFramedTransport) WriteString(s string) (n int, err error) {
 
 func (p *TFramedTransport) Flush(ctx context.Context) error {
 	size := p.writeBuf.Len()
-	if uint64(size) > uint64(math.MaxUint32) {
-		return NewTTransportException(UNKNOWN_TRANSPORT_EXCEPTION, fmt.Sprintf("frame too large: %d bytes exceeds uint32 max", size))
-	}
-
 	defer bufPool.put(&p.writeBuf)
+
+	// readFrame refuses a frame larger than the configured maximum, and so does
+	// a peer holding the same configuration. THeaderTransport.Flush holds the
+	// frames it writes to the same limit. A refused frame is dropped with the
+	// buffer.
+	if err := checkWriteFrameSize(p.cfg, size); err != nil {
+		return err
+	}
 	buf := p.buffer[:4]
 	binary.BigEndian.PutUint32(buf, uint32(size))
 	_, err := p.transport.Write(buf)

lib/go/thrift/framed_transport_test.go

@@ -168,3 +168,50 @@ func TestTFramedTransportEmptyFrames(t *testing.T) {
 		)
 	}
 }
+
+// Flush refuses a frame that a TFramedTransport holding the same configuration
+// refuses to read, before writing any of it, and the transport stays usable.
+// It refuses it the way THeaderTransport.Flush refuses one, through the check
+// the two share.
+func TestTFramedTransportFlushFrameSizeLimit(t *testing.T) {
+	const limit = 1024
+	conf := &TConfiguration{MaxFrameSize: limit}
+	ctx := context.Background()
+
+	t.Run("at-limit", func(t *testing.T) {
+		buf := NewTMemoryBuffer()
+		writer := NewTFramedTransportConf(buf, conf)
+		payload := bytes.Repeat([]byte("x"), limit)
+		writer.Write(payload)
+		if err := writer.Flush(ctx); err != nil {
+			t.Fatalf("Flush refused a frame of exactly %d bytes: %v", limit, err)
+		}
+		reader := NewTFramedTransportConf(buf, conf)
+		read := make([]byte, limit)
+		if _, err := io.ReadFull(reader, read); err != nil {
+			t.Fatalf("reading the frame back: %v", err)
+		}
+		if !bytes.Equal(read, payload) {
+			t.Error("payload read back differs from the payload written")
+		}
+	})
+
+	t.Run("over-limit", func(t *testing.T) {
+		out := &flushCountingTransport{in: bytes.NewReader(nil)}
+		writer := NewTFramedTransportConf(out, conf)
+		writer.Write(bytes.Repeat([]byte("x"), limit+1))
+		requireFlushSizeLimit(t, writer.Flush(ctx))
+		if out.written != 0 {
+			t.Errorf("Flush wrote %d bytes of a frame it refused, want 0", out.written)
+		}
+
+		// The refused frame is dropped, so the next one goes out on its own.
+		writer.Write([]byte("next"))
+		if err := writer.Flush(ctx); err != nil {
+			t.Fatalf("Flush after a refused frame: %v", err)
+		}
+		if want := 4 + len("next"); out.written != want {
+			t.Errorf("Flush after a refused frame wrote %d bytes, want %d", out.written, want)
+		}
+	})
+}

lib/go/thrift/header_transport.go

@@ -628,12 +628,13 @@ func (t *THeaderTransport) Write(p []byte) (int, error) {
 	return t.writeBuffer.Write(p)
 }
 
-// checkWriteFrameSize refuses a frame of size bytes that ReadFrame, holding the
-// same configuration, would refuse. THeaderMaxFrameSize is below the largest
-// length the frame's 32-bit length word can carry, so a frame that passes is
-// written with its true length.
-func (t *THeaderTransport) checkWriteFrameSize(size int) error {
-	if uint64(size) > uint64(THeaderMaxFrameSize) || int64(size) > int64(t.cfg.GetMaxFrameSize()) {
+// checkWriteFrameSize refuses a frame of size bytes that a reader holding cfg
+// would refuse. THeaderMaxFrameSize is below the largest length the frame's
+// 32-bit length word can carry, so a frame that passes is written with its
+// true length. TFramedTransport.Flush writes the same length word and applies
+// the same check.
+func checkWriteFrameSize(cfg *TConfiguration, size int) error {
+	if uint64(size) > uint64(THeaderMaxFrameSize) || int64(size) > int64(cfg.GetMaxFrameSize()) {
 		return NewTProtocolExceptionWithType(
 			SIZE_LIMIT,
 			fmt.Errorf("frame too large: %d bytes", size),
@@ -733,7 +734,7 @@ func (t *THeaderTransport) Flush(ctx context.Context) error {
 		}
 
 		// First write frame length
-		if err := t.checkWriteFrameSize(payload.Len()); err != nil {
+		if err := checkWriteFrameSize(t.cfg, payload.Len()); err != nil {
 			return err
 		}
 		buf := t.buffer[:size32]
@@ -747,7 +748,7 @@ func (t *THeaderTransport) Flush(ctx context.Context) error {
 		}
 
 	case clientFramedBinary, clientFramedCompact:
-		if err := t.checkWriteFrameSize(t.writeBuffer.Len()); err != nil {
+		if err := checkWriteFrameSize(t.cfg, t.writeBuffer.Len()); err != nil {
 			return err
 		}
 		buf := t.buffer[:size32]