From fb6037fe766663aeb6928c83da58f1bc6767c0fb Mon Sep 17 00:00:00 2001 From: Francesco Canovai Date: Thu, 3 Sep 2026 16:17:27 +0200 Subject: [PATCH] feat(wal): stream real per-block server acks for faster PostgreSQL feedback Report to PostgreSQL only what the Klio server has actually durably received, instead of the client's own local write/send progress, and deliver that feedback as soon as it's available. The WAL Put RPC becomes bidirectional: the server now streams back a PutResult (write_lsn/flush_lsn) per batch of blocks it flushes, instead of a single result at the very end of the file. Both sides run a background goroutine to shuttle these acks without blocking the hot path, so the client can turn them into standby status updates for PostgreSQL as soon as they arrive, rather than only once the whole file is done. Assisted-by: Claude Co-authored-by: Leonardo Cecchi Signed-off-by: Francesco Canovai --- .../klioclient/grpcclient/common_test.go | 10 +- .../klioclient/grpcclient/connection.go | 55 ++-- .../client/klioclient/grpcclient/errors.go | 4 + .../client/klioclient/grpcclient/walclient.go | 41 ++- .../{walstreamer.go => waldownloader.go} | 21 -- .../klioclient/grpcclient/waluploader.go | 99 +++++- core/internal/client/klioclient/interfaces.go | 28 +- core/internal/client/sendwal/buffer/buffer.go | 141 ++++----- core/internal/client/sendwal/buffer/grpc.go | 123 +++++--- .../internal/client/sendwal/buffer/handler.go | 11 +- core/internal/client/sendwal/buffer/memory.go | 101 ------ .../client/sendwal/feedback_sender.go | 106 +++++++ .../client/sendwal/nonblocking_receive.go | 167 ++++++++++ core/internal/client/sendwal/receiver.go | 219 ++++++------- core/internal/grpc/klio_wal.pb.go | 38 ++- core/internal/grpc/klio_wal_grpc.pb.go | 13 +- core/internal/repository/writer.go | 3 + .../server/walserver/block_receiver.go | 137 +++++++++ .../server/walserver/feedback_sender.go | 102 ++++++ core/internal/server/walserver/upload.go | 290 ++++++++++-------- core/internal/server/walserver/upload_test.go | 26 +- core/internal/wal/feedback.go | 33 ++ core/internal/walplayer/player.go | 2 +- core/pkg/config/client.go | 26 +- core/pkg/config/client_validate.go | 6 - core/pkg/config/client_validate_test.go | 46 +-- core/pkg/config/decode_test.go | 12 +- core/proto/klio_wal.proto | 8 +- documentation/web/docs/developer/_protocol.md | 6 +- operator/internal/klioconfig/config.go | 4 +- operator/pkg/config/client.go | 26 +- 31 files changed, 1180 insertions(+), 724 deletions(-) rename core/internal/client/klioclient/grpcclient/{walstreamer.go => waldownloader.go} (84%) delete mode 100644 core/internal/client/sendwal/buffer/memory.go create mode 100644 core/internal/client/sendwal/feedback_sender.go create mode 100644 core/internal/client/sendwal/nonblocking_receive.go create mode 100644 core/internal/server/walserver/block_receiver.go create mode 100644 core/internal/server/walserver/feedback_sender.go create mode 100644 core/internal/wal/feedback.go diff --git a/core/internal/client/klioclient/grpcclient/common_test.go b/core/internal/client/klioclient/grpcclient/common_test.go index 6bfcac0e..5fc4e297 100644 --- a/core/internal/client/klioclient/grpcclient/common_test.go +++ b/core/internal/client/klioclient/grpcclient/common_test.go @@ -29,12 +29,6 @@ import ( const fakeWalContent = "deadbeef" -// WALClient is the interface that wraps the backend WAL storage. -type WALClient interface { - // StoreWAL upload a WAL file to a remote store - StoreWAL(ctx context.Context, name string, content []byte, sendToTier2 bool) error -} - type testingRepository struct { prefilledSnapshots int client *TemporaryConnection @@ -111,10 +105,10 @@ func runSnapshotLookupBenchmark(b *testing.B, repo *testingRepository) { } } -func addFakeWals(ctx context.Context, repo WALClient, start int, count int) error { +func addFakeWals(ctx context.Context, repo *TemporaryConnection, start int, count int) error { for i := range count { walName := fmt.Sprintf("%024X", start+i) - err := repo.StoreWAL(ctx, walName, []byte(fakeWalContent), false) + err := repo.UploadFile(ctx, walName, []byte(fakeWalContent), false) if err != nil { return fmt.Errorf("while generating fake wal: %w", err) } diff --git a/core/internal/client/klioclient/grpcclient/connection.go b/core/internal/client/klioclient/grpcclient/connection.go index 25064d94..2e180b99 100644 --- a/core/internal/client/klioclient/grpcclient/connection.go +++ b/core/internal/client/klioclient/grpcclient/connection.go @@ -30,43 +30,48 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "github.com/cloudnative-pg/klio/core/internal/client/klioclient" klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" "github.com/cloudnative-pg/klio/core/internal/wal" "github.com/cloudnative-pg/klio/core/pkg/config" ) -type grpcWALStream struct { - innerStream klioGRPC.WAL_PutClient - segmentSize uint64 - clusterName string - sentBytes uint64 - walName string - sendToTier2 bool +// Connection represents a connection to a Klio server. +type Connection struct { + klioGRPC.WALClient + + clientConfig *config.ClientConfig + grpcConnection *grpc.ClientConn } -// Close implements common.WALStream. -func (g *grpcWALStream) Close(_ context.Context) error { - result, err := g.innerStream.CloseAndRecv() +// StoreWALStreaming implements the WAL streaming service. +// +//nolint:ireturn +func (c *Connection) StoreWALStreaming( + ctx context.Context, + name string, + segmentSize uint64, + sendToTier2 bool, + walStartLSN uint64, + feedbackChannel chan<- wal.Feedback, +) (klioclient.WALUploader, error) { + stream, err := c.Put(ctx) if err != nil { - return fmt.Errorf("while flushing WAL file: %w", err) + return nil, fmt.Errorf("while starting uploading a WAL file: %w", err) } - if result.GetWrittenSize() != g.sentBytes { - return &IncompleteWALFileError{ - uploadedSize: result.GetWrittenSize(), - expectedSize: g.sentBytes, - } + g := &grpcWALStream{ + innerStream: stream, + segmentSize: segmentSize, + clusterName: c.clientConfig.ClusterName, + walName: name, + sendToTier2: sendToTier2, + walStartLSN: walStartLSN, + feedbackChannel: feedbackChannel, } + g.startFeedbackReader() - return nil -} - -// Connection represents a connection to a Klio server. -type Connection struct { - klioGRPC.WALClient - - clientConfig *config.ClientConfig - grpcConnection *grpc.ClientConn + return g, nil } // Connect opens a connection to a Klio server. diff --git a/core/internal/client/klioclient/grpcclient/errors.go b/core/internal/client/klioclient/grpcclient/errors.go index ae0fa9eb..9184c399 100644 --- a/core/internal/client/klioclient/grpcclient/errors.go +++ b/core/internal/client/klioclient/grpcclient/errors.go @@ -27,6 +27,10 @@ import ( // ErrInconsistentCertificate is raised when the server certificate cannot be parsed. var ErrInconsistentCertificate = errors.New("inconsistent server certificate (parsing)") +// ErrNoResultReceived is raised when the server closes the WAL upload stream +// without sending a result. +var ErrNoResultReceived = errors.New("server closed stream without sending a result") + // IncompleteWALFileError is raised when a WAL file has been uploaded incompletely. type IncompleteWALFileError struct { uploadedSize uint64 diff --git a/core/internal/client/klioclient/grpcclient/walclient.go b/core/internal/client/klioclient/grpcclient/walclient.go index 388280fc..63c24a02 100644 --- a/core/internal/client/klioclient/grpcclient/walclient.go +++ b/core/internal/client/klioclient/grpcclient/walclient.go @@ -29,21 +29,21 @@ import ( klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" ) -// StoreWAL uploads a WAL in the WAL server -// Important: this function uploads a full WAL file. -func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte, sendToTier2 bool) error { +// UploadFile uploads a file to the Klio server. +// No feedback is expected from the server. +func (c *Connection) UploadFile(ctx context.Context, name string, content []byte, sendToTier2 bool) error { stream, err := c.Put(ctx) if err != nil { - return fmt.Errorf("while starting uploading a WAL file: %w", err) + return fmt.Errorf("while starting uploading a file: %w", err) } walReader := bytes.NewBuffer(content) - buffer := make([]byte, 4096) + for { readBytes, readError := walReader.Read(buffer) if readError != nil && !errors.Is(readError, io.EOF) { - return fmt.Errorf("error while reading WAL (reading from buffer): %w", readError) + return fmt.Errorf("error while reading file (reading from buffer): %w", readError) } if err := stream.Send(&klioGRPC.PutRequest{ @@ -52,8 +52,20 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte, SegmentSize: uint64(len(content)), WalBlock: buffer[:readBytes], SendToTier2: sendToTier2, + + // We're not interested to the feedback from the Klio server, + // so we just start this WAL file from LSN zero. + WalStartLsn: 0, }); err != nil { - return fmt.Errorf("error while sending WAL block (sending via GRPC): %w", err) + return fmt.Errorf("error while sending file block (sending via GRPC): %w", err) + } + + _, err := stream.Recv() + if errors.Is(err, io.EOF) { + return ErrNoResultReceived + } + if err != nil { + return fmt.Errorf("while flushing WAL file: %w", err) } if errors.Is(readError, io.EOF) { @@ -61,22 +73,9 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte, } } - result, err := stream.CloseAndRecv() - if err != nil { + if err := stream.CloseSend(); err != nil { return fmt.Errorf("while flushing WAL file: %w", err) } - if result.GetWrittenSize() != uint64(len(content)) { - return &IncompleteWALFileError{ - uploadedSize: result.GetWrittenSize(), - expectedSize: uint64(len(content)), - } - } - return nil } - -// StoreHistoryFile uses the underlying GRPC connection to store a history file. -func (c *Connection) StoreHistoryFile(ctx context.Context, name string, content []byte, sendToTier2 bool) error { - return c.StoreWAL(ctx, name, content, sendToTier2) -} diff --git a/core/internal/client/klioclient/grpcclient/walstreamer.go b/core/internal/client/klioclient/grpcclient/waldownloader.go similarity index 84% rename from core/internal/client/klioclient/grpcclient/walstreamer.go rename to core/internal/client/klioclient/grpcclient/waldownloader.go index 327afb23..9a7aaa4d 100644 --- a/core/internal/client/klioclient/grpcclient/walstreamer.go +++ b/core/internal/client/klioclient/grpcclient/waldownloader.go @@ -34,27 +34,6 @@ import ( klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" ) -// StoreWALStreaming implements the WAL streaming service. -func (c *Connection) StoreWALStreaming( - ctx context.Context, - name string, - segmentSize uint64, - sendToTier2 bool, -) (*klioclient.WALUploader, error) { - stream, err := c.Put(ctx) - if err != nil { - return nil, fmt.Errorf("while starting uploading a WAL file: %w", err) - } - - return klioclient.NewWALUploader(&grpcWALStream{ - innerStream: stream, - segmentSize: segmentSize, - clusterName: c.clientConfig.ClusterName, - walName: name, - sendToTier2: sendToTier2, - }), nil -} - // GetWALStreaming get a WAL from a remote connection. func (c *Connection) GetWALStreaming(ctx context.Context, walName string, out io.Writer) error { //nolint:cyclop client, err := c.Get(ctx, &klioGRPC.GetRequest{ diff --git a/core/internal/client/klioclient/grpcclient/waluploader.go b/core/internal/client/klioclient/grpcclient/waluploader.go index 478b9bb7..84789889 100644 --- a/core/internal/client/klioclient/grpcclient/waluploader.go +++ b/core/internal/client/klioclient/grpcclient/waluploader.go @@ -21,25 +21,67 @@ package grpcclient import ( "context" + "errors" "fmt" + "io" "time" klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" "github.com/cloudnative-pg/klio/core/internal/opentelemetry" + "github.com/cloudnative-pg/klio/core/internal/wal" ) -// SendBlock implements common.WALUploaderImpl. +// grpcWALStream uploads a WAL file to a Klio server over a single +// bidirectional gRPC stream. +// +// A background goroutine reads the acknowledgments and streams them +// to feedbackChannel. +type grpcWALStream struct { + innerStream klioGRPC.WAL_PutClient + segmentSize uint64 + clusterName string + sentBytes uint64 + walName string + walStartLSN uint64 + sendToTier2 bool + + feedbackChannel chan<- wal.Feedback + + // ackErr and ackDone follow the same pattern as messageReceiver.err in + // nonblocking_receive.go: ackErr is written at most once, by the + // background reader, always before it closes ackDone. The Go memory + // model guarantees that observing ackDone closed happens after that + // write, so reading ackErr once ackDone is (or is seen to be) closed is + // safe without any extra synchronization. + ackErr error + ackDone chan struct{} +} + +// SendBlock implements klioclient.WALUploader. It pipelines: the block is +// handed to gRPC and SendBlock returns without waiting for the server's +// per-block acknowledgment. The background goroutine started by +// startFeedbackReader pushes each acknowledgment to feedbackChannel as it +// arrives, so the caller learns how much of what has been sent is actually +// confirmed durable from there. func (g *grpcWALStream) SendBlock(ctx context.Context, block []byte) error { + if err := g.ackedErr(); err != nil { + return err + } + sendStart := time.Now() + err := g.innerStream.Send(&klioGRPC.PutRequest{ ClusterName: g.clusterName, WalName: g.walName, SegmentSize: g.segmentSize, WalBlock: block, SendToTier2: g.sendToTier2, + WalStartLsn: g.walStartLSN, }) - opentelemetry.RecordDuration(ctx, opentelemetry.ClientWal.BlockDuration, time.Since(sendStart), err, + sendDuration := time.Since(sendStart) + + opentelemetry.RecordDuration(ctx, opentelemetry.ClientWal.BlockDuration, sendDuration, err, opentelemetry.AttributeKeyClusterName.Of(g.clusterName), opentelemetry.PathPut.Attribute(), opentelemetry.StageSend.Attribute()) @@ -51,3 +93,56 @@ func (g *grpcWALStream) SendBlock(ctx context.Context, block []byte) error { return nil } + +// Close stops sending on the stream and waits for the background ack +// reader to observe the end of the stream, surfacing any error it recorded. +func (g *grpcWALStream) Close(_ context.Context) error { + if err := g.innerStream.CloseSend(); err != nil { + return err //nolint:wrapcheck + } + + <-g.ackDone + + return g.ackErr +} + +// startFeedbackReader starts the background goroutine that drains the server's +// per-block acknowledgments. +func (g *grpcWALStream) startFeedbackReader() { + g.ackDone = make(chan struct{}) + + go func() { + defer close(g.ackDone) + + for { + result, err := g.innerStream.Recv() + if err != nil { + if !errors.Is(err, io.EOF) { + g.ackErr = fmt.Errorf("error while receiving WAL block ack: %w", err) + } + + return + } + + if g.feedbackChannel != nil { + g.feedbackChannel <- wal.Feedback{ + FlushLSN: result.GetFlushLsn(), + WriteLSN: result.GetWriteLsn(), + ReplayLSN: result.GetFlushLsn(), + } + } + } + }() +} + +// ackedErr returns the error recorded by the background ack reader once it +// has observed the end of the stream, so SendBlock can fail fast instead of +// sending into a stream already known to be dead. +func (g *grpcWALStream) ackedErr() error { + select { + case <-g.ackDone: + return g.ackErr + default: + return nil + } +} diff --git a/core/internal/client/klioclient/interfaces.go b/core/internal/client/klioclient/interfaces.go index 6c4642f4..43028c63 100644 --- a/core/internal/client/klioclient/interfaces.go +++ b/core/internal/client/klioclient/interfaces.go @@ -97,34 +97,12 @@ type Client interface { Close(ctx context.Context) } -// WALUploaderImpl is the underlying implementation of a WAL +// WALUploader is the underlying implementation of a WAL // uploader. -type WALUploaderImpl interface { - // SendBlock sends a WAL Block +type WALUploader interface { + // SendBlock sends a WAL Block. SendBlock(ctx context.Context, block []byte) error // Close closes the WAL streaming session Close(ctx context.Context) error } - -// WALUploader allows the user the upload a WAL file to a remote store, block by block. -type WALUploader struct { - impl WALUploaderImpl -} - -// NewWALUploader creates a WAL uploader given the underlying implementation. -func NewWALUploader(impl WALUploaderImpl) *WALUploader { - return &WALUploader{ - impl: impl, - } -} - -// SendBlock sends a WAL Block. -func (u *WALUploader) SendBlock(ctx context.Context, block []byte) error { - return u.impl.SendBlock(ctx, block) //nolint:wrapcheck -} - -// Close closes the WAL streaming session. -func (u *WALUploader) Close(ctx context.Context) error { - return u.impl.Close(ctx) //nolint:wrapcheck -} diff --git a/core/internal/client/sendwal/buffer/buffer.go b/core/internal/client/sendwal/buffer/buffer.go index 5be1513e..1e030eb1 100644 --- a/core/internal/client/sendwal/buffer/buffer.go +++ b/core/internal/client/sendwal/buffer/buffer.go @@ -26,6 +26,7 @@ import ( "github.com/cloudnative-pg/machinery/pkg/log" "github.com/cloudnative-pg/machinery/pkg/types" + "github.com/jackc/pglogrepl" ) // maximumBufferSizeFactor allows configuring the higher limit of memory @@ -38,20 +39,20 @@ type Data struct { segmentSize uint64 tli int - handler Handler + walHandler WALHandler + + localWriteLSN uint64 - writeLSN uint64 - flushLSN uint64 buffer *bytes.Buffer bufferSize int } // New creates a new WAL buffer. -func New(tli int, walSegmentSize uint64, handler Handler, bufferSize int) *Data { +func New(tli int, walSegmentSize uint64, handler WALHandler, bufferSize int) *Data { result := &Data{ segmentSize: walSegmentSize, tli: tli, - handler: handler, + walHandler: handler, bufferSize: bufferSize, } @@ -63,32 +64,23 @@ func New(tli int, walSegmentSize uint64, handler Handler, bufferSize int) *Data // ProcessWALData processes a WAL message from PG // //nolint:cyclop -func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types.LSN) error { - contextLogger := log.FromContext(ctx) - +func (wal *Data) ProcessWALData(ctx context.Context, data []byte, blockpos pglogrepl.LSN) error { // This implementation is largely based on src/bin/pg_basebackup/receivelog.c // [ProcessXLogDataMsg] //nolint:lll // See: https://github.com/postgres/postgres/blob/00f4c2959d631c7851da21a512885d1deab28649/src/bin/pg_basebackup/receivelog.c#L1039 - contextLogger.Debug("Process WAL Data", "lenData", len(data), "startWAL", startWAL) + xlogoff := uint64(blockpos) % wal.segmentSize - blockpos, err := startWAL.Parse() - if err != nil { - return fmt.Errorf("while parsing WAL data start (pos): %w", err) - } - - xlogoff := blockpos % wal.segmentSize - - if !wal.handler.HasWALFileOpened() { + if !wal.walHandler.HasWALFileOpened() { if xlogoff != 0 { // No file open yet return &UnopenedFileForWALError{offset: xlogoff} } } else { // More data in existing segment - currentOffset := wal.writeLSN % wal.segmentSize + currentOffset := wal.localWriteLSN % wal.segmentSize if currentOffset != xlogoff { return &UnexpectedWalDataOffsetError{ offset: xlogoff, @@ -110,24 +102,24 @@ func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types bytesToWrite = bytesLeft } - if !wal.handler.HasWALFileOpened() { - if err := wal.openWALPos(ctx, blockpos); err != nil { + if !wal.walHandler.HasWALFileOpened() { + if err := wal.openWALPos(ctx, uint64(blockpos)); err != nil { return err } } - if err := wal.writeToWALFile(ctx, data[bytesWritten:bytesWritten+bytesToWrite]); err != nil { + if err := wal.writeToWALFile(data[bytesWritten : bytesWritten+bytesToWrite]); err != nil { return fmt.Errorf("while writing to WAL handler: %w", err) } bytesWritten += bytesToWrite bytesLeft -= bytesToWrite - blockpos += bytesToWrite + blockpos += pglogrepl.LSN(bytesToWrite) xlogoff += bytesToWrite // Did we reach the end of a WAL segment? - if currentOffset := wal.writeLSN % wal.segmentSize; currentOffset == 0 { - if err := wal.closeCurrentWAL(ctx); err != nil { + if currentOffset := wal.localWriteLSN % wal.segmentSize; currentOffset == 0 { + if err := wal.CloseCurrentWAL(ctx); err != nil { return err } @@ -138,63 +130,14 @@ func (wal *Data) ProcessWALData(ctx context.Context, data []byte, startWAL types return nil } -// FlushLSN gets the latest LSN that was flushed down to the Klio server. -func (wal *Data) FlushLSN() uint64 { - return wal.flushLSN -} - -// WriteLSN gets the latest LSN that was written into the memory. -func (wal *Data) WriteLSN() uint64 { - return wal.writeLSN -} - -// Flush flushes the buffer to the Klio server connection. +// Flush writes any buffered data to the currently open WAL file. It is a +// no-op when no WAL file is open or the buffer is empty. func (wal *Data) Flush(ctx context.Context) error { - return wal.flushInternal(ctx) -} - -func (wal *Data) newBuffer() *bytes.Buffer { - return bytes.NewBuffer(make([]byte, 0, wal.bufferSize)) -} - -func (wal *Data) openWALPos(ctx context.Context, blockpos uint64) error { - contextLogger := log.FromContext(ctx) - contextLogger.Info("Opening WAL file", "blockpos", types.Int64ToLSN(blockpos)) - - if err := wal.handler.OpenWAL(ctx, blockpos); err != nil { - return err //nolint:wrapcheck - } - - wal.writeLSN = blockpos - wal.flushLSN = blockpos - - return nil -} - -func (wal *Data) writeToWALFile(ctx context.Context, data []byte) error { - if _, err := wal.buffer.Write(data); err != nil { - return fmt.Errorf("while writing to buffer: %w", err) - } - - wal.writeLSN += uint64(len(data)) - - if wal.buffer.Len() >= wal.bufferSize { - return wal.Flush(ctx) - } - - return nil -} - -func (wal *Data) flushInternal(ctx context.Context) error { - contextLogger := log.FromContext(ctx) - - if wal.handler == nil || !wal.handler.HasWALFileOpened() || wal.buffer.Len() == 0 { + if wal.walHandler == nil || !wal.walHandler.HasWALFileOpened() || wal.buffer.Len() == 0 { return nil } - contextLogger.Debug("Writing block", - "blockpos", types.Int64ToLSN(wal.writeLSN), "blocksize", wal.buffer.Len()) - _, err := wal.handler.Write(ctx, wal.buffer.Bytes()) + _, err := wal.walHandler.Write(ctx, wal.buffer.Bytes()) if err != nil { return fmt.Errorf("while writing to WAL handler: %w", err) } @@ -207,22 +150,58 @@ func (wal *Data) flushInternal(ctx context.Context) error { wal.buffer = wal.newBuffer() } - wal.flushLSN = wal.writeLSN - return nil } -func (wal *Data) closeCurrentWAL(ctx context.Context) error { +// CloseCurrentWAL flushes and closes the currently open WAL file. It is a +// no-op when no WAL file is open. +func (wal *Data) CloseCurrentWAL(ctx context.Context) error { contextLogger := log.FromContext(ctx) contextLogger.Debug("Closing WAL file") + if !wal.walHandler.HasWALFileOpened() { + return nil + } if err := wal.Flush(ctx); err != nil { return fmt.Errorf("while flushing WAL handler: %w", err) } - if err := wal.handler.CloseWAL(ctx); err != nil { + if err := wal.walHandler.CloseWAL(ctx); err != nil { return fmt.Errorf("while closing current WAL file: %w", err) } return nil } + +func (wal *Data) newBuffer() *bytes.Buffer { + return bytes.NewBuffer(make([]byte, 0, wal.bufferSize)) +} + +func (wal *Data) openWALPos(ctx context.Context, blockpos uint64) error { + contextLogger := log.FromContext(ctx) + contextLogger.Info("Opening WAL file", "blockpos", types.Int64ToLSN(blockpos)) + + if err := wal.walHandler.OpenWAL(ctx, blockpos); err != nil { + return err //nolint:wrapcheck + } + + wal.localWriteLSN = blockpos + + return nil +} + +// writeToWALFile appends data to the in-memory buffer. It does not flush: +// like a PostgreSQL walreceiver, which writes whatever it received and +// flushes once per receive cycle rather than accumulating to a target size, +// flushing here is left entirely to the caller (once per drained batch of +// messages, or at segment close) so a block sent to the Klio server tracks +// how much WAL actually arrived at once, not an unrelated size threshold. +func (wal *Data) writeToWALFile(data []byte) error { + if _, err := wal.buffer.Write(data); err != nil { + return fmt.Errorf("while writing to buffer: %w", err) + } + + wal.localWriteLSN += uint64(len(data)) + + return nil +} diff --git a/core/internal/client/sendwal/buffer/grpc.go b/core/internal/client/sendwal/buffer/grpc.go index 6cf88658..b28ab35f 100644 --- a/core/internal/client/sendwal/buffer/grpc.go +++ b/core/internal/client/sendwal/buffer/grpc.go @@ -28,15 +28,16 @@ import ( "github.com/cloudnative-pg/klio/core/internal/client/klioclient" "github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient" + "github.com/cloudnative-pg/klio/core/internal/wal" ) -// KlioClientStreamingHandler is a handler that streams directly to a +// KlioClientTimelineHandler is a handler that streams directly to a // Klio server. -type KlioClientStreamingHandler struct { +type KlioClientTimelineHandler struct { conn *grpcclient.Connection - stream klioclient.WALUploaderImpl - offset uint64 + stream klioclient.WALUploader + feedbackChannel chan<- wal.Feedback sendToTier2 bool @@ -45,76 +46,114 @@ type KlioClientStreamingHandler struct { currentWALFile string } -// NewKlioClientHandler creates a new klio handler. -func NewKlioClientHandler( +// NewKlioClientTimelineHandler creates a new klio handler. +func NewKlioClientTimelineHandler( tli int, segmentSize uint64, conn *grpcclient.Connection, sendToTier2 bool, -) *KlioClientStreamingHandler { - return &KlioClientStreamingHandler{ - conn: conn, - tli: tli, - segmentSize: segmentSize, - stream: nil, - sendToTier2: sendToTier2, + feedbackChannel chan<- wal.Feedback, +) *KlioClientTimelineHandler { + return &KlioClientTimelineHandler{ + conn: conn, + tli: tli, + segmentSize: segmentSize, + stream: nil, + sendToTier2: sendToTier2, + feedbackChannel: feedbackChannel, } } -// OpenWAL implements the Handler interface. -func (wal *KlioClientStreamingHandler) OpenWAL(ctx context.Context, blockpos uint64) error { - currentWALFile, err := types.Int64ToLSN(blockpos).WALFileName(wal.tli, wal.segmentSize) +// OpenWAL implements the WALHandler interface. +func (timelineHandler *KlioClientTimelineHandler) OpenWAL(ctx context.Context, blockpos uint64) error { + currentWALFile, err := types.Int64ToLSN(blockpos).WALFileName(timelineHandler.tli, timelineHandler.segmentSize) if err != nil { return fmt.Errorf("while creating WAL file name (pos %v): %w", blockpos, err) } - wal.offset = 0 - wal.currentWALFile = currentWALFile + timelineHandler.currentWALFile = currentWALFile + + // We are starting a new WAL: report progress up to its start position + // right away, rather than waiting for the Klio server to ack any of its + // blocks. Otherwise PG would see our reported LSNs stall at the end of + // the previous file for as long as the new file's first ack takes to + // arrive, which can look like the standby has stopped progressing. + timelineHandler.feedbackChannel <- wal.Feedback{ + FlushLSN: blockpos, + WriteLSN: blockpos, + ReplayLSN: blockpos, + } - stream, err := wal.conn.StoreWALStreaming(ctx, wal.currentWALFile, wal.segmentSize, wal.sendToTier2) + stream, err := timelineHandler.conn.StoreWALStreaming( + ctx, + timelineHandler.currentWALFile, + timelineHandler.segmentSize, + timelineHandler.sendToTier2, + blockpos, + timelineHandler.feedbackChannel, + ) if err != nil { return fmt.Errorf("while starting WAL file streaming (pos %v): %w", blockpos, err) } - wal.stream = stream + timelineHandler.stream = stream return nil } -// HasWALFileOpened implements the Handler interface. -func (wal *KlioClientStreamingHandler) HasWALFileOpened() bool { - return wal.currentWALFile != "" +// HasWALFileOpened implements the WALHandler interface. +func (timelineHandler *KlioClientTimelineHandler) HasWALFileOpened() bool { + return timelineHandler.currentWALFile != "" } -// CloseWAL implements the Handler interface. -func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error { +// CloseWAL implements the WALHandler interface. +func (timelineHandler *KlioClientTimelineHandler) CloseWAL(ctx context.Context) error { contextLogger := log.FromContext(ctx) - contextLogger.Debug("Closing WAL File", "walFileName", wal.currentWALFile) + contextLogger.Debug("Closing WAL File", "walFileName", timelineHandler.currentWALFile) - if err := wal.stream.Close(ctx); err != nil { + if err := timelineHandler.stream.Close(ctx); err != nil { return err //nolint:wrapcheck } - wal.currentWALFile = "" - wal.stream = nil + timelineHandler.currentWALFile = "" + timelineHandler.stream = nil return nil } -// CurrentOffset implements the Handler interface. -func (wal *KlioClientStreamingHandler) CurrentOffset() (uint64, error) { - return wal.offset, nil -} - -// Write implements the Handler interface. -func (wal *KlioClientStreamingHandler) Write(ctx context.Context, block []byte) (int, error) { - err := wal.stream.SendBlock(ctx, block) - if err != nil { - return 0, err //nolint:wrapcheck +// maxWriteChunkBytes caps how much of a Write call is handed to a single +// SendBlock call. buffer.Data no longer caps how much it accumulates before +// flushing (it flushes at whatever granularity WAL was received in, like a +// PostgreSQL walreceiver), so a single Write here may carry more than one +// gRPC message is allowed to hold - the server rejects anything over +// wal.MaxBlockSizeBytes (8 MiB); this stays safely under that. +const maxWriteChunkBytes = 4 * 1024 * 1024 + +// Write implements the WALHandler interface. It splits block into chunks of +// at most maxWriteChunkBytes, each sent with its own SendBlock call, so that +// a single Write never hands gRPC a message over the server's accepted size +// (see maxWriteChunkBytes). As a side effect, acks - delivered asynchronously +// on feedbackChannel by grpcWALStream's background reader - can arrive +// progressively across chunks rather than only once at the very end, though +// the server may still coalesce several chunks into a single flush and +// acknowledgment (see blockReceiver.Drain). +func (timelineHandler *KlioClientTimelineHandler) Write(ctx context.Context, block []byte) (int, error) { + written := 0 + + for len(block) > 0 { + chunk := block + if len(chunk) > maxWriteChunkBytes { + chunk = chunk[:maxWriteChunkBytes] + } + + if err := timelineHandler.stream.SendBlock(ctx, chunk); err != nil { + return written, err //nolint:wrapcheck + } + + written += len(chunk) + block = block[len(chunk):] } - wal.offset += uint64(len(block)) - - return len(block), nil + return written, nil } diff --git a/core/internal/client/sendwal/buffer/handler.go b/core/internal/client/sendwal/buffer/handler.go index d17859a4..6e3d10de 100644 --- a/core/internal/client/sendwal/buffer/handler.go +++ b/core/internal/client/sendwal/buffer/handler.go @@ -19,11 +19,13 @@ SPDX-License-Identifier: Apache-2.0 package buffer -import "context" +import ( + "context" +) -// Handler is the interface used to process WAL data. +// WALHandler is the interface used to process WAL data. // This is vastly modeled around the pg_basebackup codebase. -type Handler interface { +type WALHandler interface { // HasWALFileOpened Checks whether there is a WAL file transmission opened HasWALFileOpened() bool @@ -34,9 +36,6 @@ type Handler interface { // CloseWAL closes a WAL file CloseWAL(ctx context.Context) error - // CurrentOffset returns the current offset in the WAL file - CurrentOffset() (uint64, error) - // Write writes data in the current WAL file Write(ctx context.Context, p []byte) (n int, err error) } diff --git a/core/internal/client/sendwal/buffer/memory.go b/core/internal/client/sendwal/buffer/memory.go deleted file mode 100644 index 3021af04..00000000 --- a/core/internal/client/sendwal/buffer/memory.go +++ /dev/null @@ -1,101 +0,0 @@ -/* -Copyright © contributors to CloudNativePG, established as -CloudNativePG a Series of LF Projects, LLC. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. - -SPDX-License-Identifier: Apache-2.0 -*/ - -package buffer - -import ( - "bytes" - "context" - "fmt" - - "github.com/ccoveille/go-safecast/v2" - "github.com/cloudnative-pg/machinery/pkg/log" - "github.com/cloudnative-pg/machinery/pkg/types" -) - -// Flusher is the type of functions that are called -// to write a WAL file. -type Flusher func(walName string, data []byte) error - -// MemBufferHandler is the handler of WAL files that writes in a memory buffer -// and, when the WAL is completed, flushes it via a Flusher function. -type MemBufferHandler struct { - currentWALFile string - buffer bytes.Buffer - logger log.Logger - flusher Flusher - - tli int - segmentSize uint64 -} - -// NewMemBufferHandler creates a new memory buffer handler. -func NewMemBufferHandler(logger log.Logger, tli int, segmentSize uint64, flusher Flusher) *MemBufferHandler { - return &MemBufferHandler{ - currentWALFile: "", - buffer: *bytes.NewBuffer(make([]byte, 0, segmentSize)), - logger: logger, - flusher: flusher, - tli: tli, - segmentSize: segmentSize, - } -} - -// HasWALFileOpened implements the Handler interface. -func (wal *MemBufferHandler) HasWALFileOpened() bool { - return wal.currentWALFile != "" -} - -// OpenWAL implements the Handler interface. -func (wal *MemBufferHandler) OpenWAL(_ context.Context, blockpos uint64) error { - var err error - - wal.currentWALFile, err = types.Int64ToLSN(blockpos).WALFileName(wal.tli, wal.segmentSize) - if err != nil { - return fmt.Errorf("while creating WAL file name (pos %v): %w", blockpos, err) - } - wal.buffer.Reset() - - wal.logger.Debug("Opening WAL File", "walFileName", wal.currentWALFile) - - return nil -} - -// CloseWAL implements the Handler interface. -func (wal *MemBufferHandler) CloseWAL(_ context.Context) error { - wal.logger.Debug("Closing WAL File", "walFileName", wal.currentWALFile) - if err := wal.flusher(wal.currentWALFile, wal.buffer.Bytes()); err != nil { - return err - } - - wal.currentWALFile = "" - wal.buffer.Reset() - - return nil -} - -// CurrentOffset implements the Handler interface. -func (wal *MemBufferHandler) CurrentOffset() (uint64, error) { - return safecast.Convert[uint64](wal.buffer.Len()) -} - -// Write implements the Handler interface. -func (wal *MemBufferHandler) Write(_ context.Context, p []byte) (int, error) { - return wal.buffer.Write(p) //nolint:wrapcheck -} diff --git a/core/internal/client/sendwal/feedback_sender.go b/core/internal/client/sendwal/feedback_sender.go new file mode 100644 index 00000000..e3cde0b0 --- /dev/null +++ b/core/internal/client/sendwal/feedback_sender.go @@ -0,0 +1,106 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package sendwal + +import ( + "context" + "time" + + "github.com/cloudnative-pg/machinery/pkg/log" + "github.com/jackc/pglogrepl" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/cloudnative-pg/klio/core/internal/wal" +) + +// feedbackSender decouples sending standby status updates to PostgreSQL from +// the rate at which new feedback values arrive and reply requests come in. +type feedbackSender struct { + feedbackChannel <-chan wal.Feedback + conn *pgconn.PgConn + latest wal.Feedback + + wake chan struct{} + done chan struct{} +} + +// newFeedbackSender starts the background send loop for conn. Stop must be +// called once the sender is no longer needed, to release the goroutine. +func newFeedbackSender( + ctx context.Context, + conn *pgconn.PgConn, + feedbackChannel <-chan wal.Feedback, +) *feedbackSender { + s := &feedbackSender{ + wake: make(chan struct{}, 1), + done: make(chan struct{}), + conn: conn, + feedbackChannel: feedbackChannel, + } + + go s.run(ctx) + + return s +} + +// Send queues update to be sent. +func (s *feedbackSender) Send() { + select { + case s.wake <- struct{}{}: + default: + } +} + +// Stop waits for the background goroutine to exit. It does not by itself +// make that happen: the caller must close feedbackChannel first. Once Stop returns, +// nothing is writing to conn on this sender's behalf. Safe to call more than once. +func (s *feedbackSender) Stop() { + <-s.done +} + +func (s *feedbackSender) run(ctx context.Context) { + defer close(s.done) + + for { + select { + case latest, ok := <-s.feedbackChannel: + if !ok { + return + } + + s.latest = latest + s.sendLatest(ctx) + case <-s.wake: + s.sendLatest(ctx) + } + } +} + +func (s *feedbackSender) sendLatest(ctx context.Context) { + msg := pglogrepl.StandbyStatusUpdate{ + WALWritePosition: pglogrepl.LSN(s.latest.WriteLSN), + WALFlushPosition: pglogrepl.LSN(s.latest.FlushLSN), + WALApplyPosition: pglogrepl.LSN(s.latest.ReplayLSN), + ClientTime: time.Now(), + } + if err := pglogrepl.SendStandbyStatusUpdate(ctx, s.conn, msg); err != nil { + log.FromContext(ctx).Error(err, "Failed to send standby status update, skipping") + } +} diff --git a/core/internal/client/sendwal/nonblocking_receive.go b/core/internal/client/sendwal/nonblocking_receive.go new file mode 100644 index 00000000..43587195 --- /dev/null +++ b/core/internal/client/sendwal/nonblocking_receive.go @@ -0,0 +1,167 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package sendwal + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgproto3" +) + +// ErrMessageReceiverClosed is a fallback returned when the background +// receive loop of a messageReceiver has exited without recording an error, +// which should not normally happen. +var ErrMessageReceiverClosed = errors.New("message receiver closed") + +// cloneMessage returns a copy of msg that is safe to keep after the next +// call to conn.ReceiveMessage. +// +// pgproto3's Frontend reads messages into a shared, reused buffer: the +// byte slices it hands back (e.g. CopyData.Data) are only valid until the +// next Receive call. The background loop in messageReceiver queues several +// messages before any of them are processed, so without copying here, an +// earlier queued message's payload gets silently overwritten by a later +// read of the same buffer. +// +//nolint:ireturn +func cloneMessage(msg pgproto3.BackendMessage) pgproto3.BackendMessage { + if cd, ok := msg.(*pgproto3.CopyData); ok { + data := make([]byte, len(cd.Data)) + copy(data, cd.Data) + + return &pgproto3.CopyData{Data: data} + } + + return msg +} + +// messageReceiver decouples receiving PostgreSQL protocol messages from the +// caller's polling cadence. A background goroutine blocks on +// conn.ReceiveMessage in a tight loop and publishes every message on a +// channel, so that draining messages already available on the wire never +// needs an artificial read deadline: it is answered instantly by however +// many items are already queued in the channel. +// +// Stop must be called before conn is read from by anything else - e.g. +// before pglogrepl.SendStandbyCopyDone, which reads directly off the same +// underlying frontend. Without it, the background goroutine keeps calling +// conn.ReceiveMessage after the caller has moved on, racing whoever reads +// from conn next for messages meant for them. +type messageReceiver struct { + ch <-chan pgproto3.BackendMessage + cancel context.CancelFunc + done chan struct{} + + // err is set by the background goroutine, once, before it closes ch. + // The Go memory model guarantees that a receive observing ch as closed + // happens after that close, so reading err after such a receive is safe + // without extra synchronization: only one goroutine ever writes it, and + // only after all sends are done. + err error +} + +// newMessageReceiver starts the background receive loop for conn. The loop, +// and the returned receiver, are only valid until ctx is done or Stop is +// called, whichever happens first; after either, ReceiveAvailable starts +// failing. +func newMessageReceiver(ctx context.Context, conn *pgconn.PgConn) *messageReceiver { + ctx, cancel := context.WithCancel(ctx) + ch := make(chan pgproto3.BackendMessage, 500) + receiver := &messageReceiver{ch: ch, cancel: cancel, done: make(chan struct{})} + + go func() { + defer close(ch) + defer close(receiver.done) + + for { + msg, err := conn.ReceiveMessage(ctx) + if err != nil { + receiver.err = err + + return + } + + select { + case ch <- cloneMessage(msg): + case <-ctx.Done(): + receiver.err = ctx.Err() + + return + } + } + }() + + return receiver +} + +// Stop cancels the background receive loop and waits for it to exit. It is +// safe to call more than once. Once Stop returns, the background goroutine +// is no longer calling conn.ReceiveMessage, so conn is safe to read from +// directly. +func (r *messageReceiver) Stop() { + r.cancel() + <-r.done +} + +// ReceiveAvailable blocks until at least one message is available, then +// drains every message that was already queued without blocking again - +// so a burst of messages already on the wire is handed to the caller +// together, letting it process them (and flush the result) as one batch +// instead of one at a time. +func (r *messageReceiver) ReceiveAvailable(ctx context.Context) ([]pgproto3.BackendMessage, error) { + select { + case msg, ok := <-r.ch: + if !ok { + return nil, r.closedErr() + } + + messages := make([]pgproto3.BackendMessage, 0, 1) + messages = append(messages, msg) + + for { + select { + case msg, ok := <-r.ch: + if !ok { + return messages, nil + } + + messages = append(messages, msg) + default: + return messages, nil + } + } + + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// closedErr returns the error recorded by the background goroutine once it +// has closed the channel, falling back to ErrMessageReceiverClosed in the +// (unexpected) case none was recorded. +func (r *messageReceiver) closedErr() error { + if r.err != nil { + return r.err + } + + return ErrMessageReceiverClosed +} diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index 245d5a68..381a318b 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -25,7 +25,6 @@ import ( "fmt" "path" "strings" - "time" "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres" "github.com/cloudnative-pg/machinery/pkg/log" @@ -42,6 +41,7 @@ import ( "github.com/cloudnative-pg/klio/core/internal/client/sendwal/infrastructure" klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc" "github.com/cloudnative-pg/klio/core/internal/opentelemetry" + "github.com/cloudnative-pg/klio/core/internal/wal" "github.com/cloudnative-pg/klio/core/pkg/config" ) @@ -63,7 +63,7 @@ func New(cfg *config.Data, logger log.Logger, client *grpcclient.Connection, sen } } -// ResetReplicationStatus reset the replication status on the server side and then +// ResetReplicationStatus resets the replication status on the server side and then // drops the Klio replication slot. func (s *Process) ResetReplicationStatus( ctx context.Context, @@ -368,7 +368,7 @@ func (s *Process) downloadHistoryFiles( continue } - if err := s.client.StoreHistoryFile(ctx, result.FileName, result.Content, s.sendToTier2); err != nil { + if err := s.client.UploadFile(ctx, result.FileName, result.Content, s.sendToTier2); err != nil { span.RecordError(err) errorList = errors.Join(errorList, err) contextLogger.Error(err, "timeline history upload failed", @@ -433,11 +433,14 @@ func (s *Process) startReplication( "timeline", timeline, ) - klioHandler := buffer.NewKlioClientHandler( + feedbackChannel := make(chan wal.Feedback, 100) + + klioHandler := buffer.NewKlioClientTimelineHandler( int(timeline), walSegmentSize, s.client, s.sendToTier2, + feedbackChannel, ) walBuffer := buffer.New( @@ -447,20 +450,11 @@ func (s *Process) startReplication( s.config.Source.BufferSize, ) - copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer) + copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer, feedbackChannel) if err != nil { return err } - if klioHandler.HasWALFileOpened() { - // If the transmission terminated but there is still a WAL file in progress, - // we close it. - // This happens when PG is shut down. - if err := klioHandler.CloseWAL(ctx); err != nil { - return fmt.Errorf("while closing the WAL file: %w", err) - } - } - // Check if the timeline has changed and restart replication if needed if copyDoneResult != nil && copyDoneResult.Timeline != timeline { contextLogger.Info( @@ -491,122 +485,107 @@ func (s *Process) manageWALStream( ctx context.Context, conn *pgconn.PgConn, buffer *buffer.Data, + feedbackChannel chan wal.Feedback, ) (*pglogrepl.CopyDoneResult, error) { contextLogger := log.FromContext(ctx) - flushDeadline := s.config.Source.FlushTimeout() - nextFlushDeadline := time.Now().Add(flushDeadline) - - feedbackDeadline := s.config.Source.StandbyMessageTimeout() - nextFeedbackDeadline := time.Now().Add(feedbackDeadline) + walReceiver := newMessageReceiver(ctx, conn) + feedbackSender := newFeedbackSender(ctx, conn, feedbackChannel) loop: for { - if time.Now().After(nextFlushDeadline) { - flushedLSN := buffer.FlushLSN() - - if err := buffer.Flush(ctx); err != nil { - contextLogger.Error(err, "Failed flush WAL data") - return nil, fmt.Errorf("while flushing WAL data: %w", err) - } - - // When flush really written something down to the Klio server, - // the FlushedLSN will be different. In that case, we want to immediately - // give feedback to the PostgreSQL server. This ultimately - // will result in updated data in pg_stat_replication. - if flushedLSN != buffer.FlushLSN() { - nextFeedbackDeadline = time.Time{} - } - - nextFlushDeadline = time.Now().Add(flushDeadline) - } - - if time.Now().After(nextFeedbackDeadline) { - // We communicate back to PostgreSQL the feedback when: - // - // 1. the feedback deadline exceeded - // 2. we received something from streaming replication - s.sendFeedback(ctx, conn, buffer) - nextFeedbackDeadline = time.Now().Add(feedbackDeadline) - } - - standbyMessageDeadlineContext, cancel := context.WithDeadline(ctx, nextFlushDeadline) - msg, err := conn.ReceiveMessage(standbyMessageDeadlineContext) - cancel() - + messages, err := walReceiver.ReceiveAvailable(ctx) if err != nil { - if pgconn.Timeout(err) { - continue - } - if errors.Is(err, context.Canceled) { - break - } contextLogger.Error(err, "receive message failed") - - break + break loop } - log.FromContext(ctx).Trace( - "Received message", - "msgType", fmt.Sprintf("%T", msg)) - - switch msg := msg.(type) { - case *pgproto3.CopyData: - switch msg.Data[0] { - case pglogrepl.PrimaryKeepaliveMessageByteID: - pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:]) - if err != nil { - contextLogger.Error(err, "parsePrimaryKeepaliveMessage failed") - continue - } - contextLogger.Debug( - "Primary Keepalive Message", - "ServerWALEnd", pkm.ServerWALEnd, - "ServerTime", pkm.ServerTime, - "ReplyRequested", pkm.ReplyRequested, - ) - - if pkm.ReplyRequested { - s.sendFeedback(ctx, conn, buffer) + for _, msg := range messages { + switch msg := msg.(type) { + case *pgproto3.CopyData: + switch msg.Data[0] { + case pglogrepl.PrimaryKeepaliveMessageByteID: + pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:]) + if err != nil { + contextLogger.Error(err, "parsePrimaryKeepaliveMessage failed") + continue + } + contextLogger.Debug( + "Primary Keepalive Message", + "ServerWALEnd", pkm.ServerWALEnd, + "ServerTime", pkm.ServerTime, + "ReplyRequested", pkm.ReplyRequested, + ) + + if pkm.ReplyRequested { + feedbackSender.Send() + } + + case pglogrepl.XLogDataByteID: + xld, err := pglogrepl.ParseXLogData(msg.Data[1:]) + if err != nil { + contextLogger.Error(err, "ParseXLogData failed") + continue + } + + err = buffer.ProcessWALData(ctx, xld.WALData, xld.WALStart) + if err != nil { + contextLogger.Error(err, "Error while processing WAL data", "lsn", xld.WALStart) + break loop + } + + default: + contextLogger.Info("Received unexpected copydata message", "msg", msg) + break loop } - case pglogrepl.XLogDataByteID: - xld, err := pglogrepl.ParseXLogData(msg.Data[1:]) - if err != nil { - contextLogger.Error(err, "ParseXLogData failed") - continue - } + case *pgproto3.CommandComplete: + contextLogger.Info("Streaming replication terminated by the backend with success") + break loop - err = buffer.ProcessWALData(ctx, xld.WALData, types.LSN(xld.WALStart.String())) - if err != nil { - contextLogger.Error(err, "Error while processing WAL data", "lsn", xld.WALStart) - - return nil, fmt.Errorf("could not process WAL data at %s: %w", xld.WALStart, err) - } - - // Force the code to communicate back to PostgreSQL the current status without waiting for - // a flush - nextFeedbackDeadline = time.Time{} + case *pgproto3.CopyDone: + contextLogger.Info("Streaming replication terminated by the backend with CopyDone") + break loop default: - contextLogger.Info("Received unexpected copydata message", "msg", msg) - return nil, NewUnexpectedCopydataMessageError(msg.Data) + contextLogger.Info("Received unexpected message", "msg", msg) + return nil, NewUnexpectedMessageError(msg) } + } - case *pgproto3.CommandComplete: - contextLogger.Info("Streaming replication terminated by the backend with success") - return nil, nil - - case *pgproto3.CopyDone: - contextLogger.Info("Streaming replication terminated by the backend with CopyDone") + // Reporting the write position and the flushed position to PG is + // handled reactively by feedbackSender (started above), driven + // directly off buffer's write-position and ack updates - not from + // here, so neither has to wait for Flush, which can block on the + // Klio server. One Flush per drained batch, rather than one per + // XLogData message, turns a burst of messages already on the wire + // into a single write to the Klio server. + if err := buffer.Flush(ctx); err != nil { + contextLogger.Error(err, "cannot flush data to Klio server") break loop - - default: - contextLogger.Info("Received unexpected message", "msg", msg) - return nil, NewUnexpectedMessageError(msg) } } + // Stop both background goroutines before touching conn directly below: + // SendStandbyCopyDone reads off the same underlying frontend, and would + // race with a still-running receiver for the messages it expects. + contextLogger.Info("Stopping WAL receiver from PostgreSQL") + walReceiver.Stop() + + // We close the WAL we're writing, and this ensures we received all the feedback. + contextLogger.Info("Closing WAL sender to Klio server") + if err := buffer.CloseCurrentWAL(ctx); err != nil { + return nil, fmt.Errorf("cannot close WAL sender to Klio server: %w", err) + } + + // The reader is stopped and we read everything in it. We can close the feedback + // channel allowing the feedback sender to stop. + close(feedbackChannel) + + // Stop the feedback sender and flush the final status. + contextLogger.Info("Stopping feedback sender to PostgreSQL") + feedbackSender.Stop() + contextLogger.Info("WAL streaming loop terminated, sending CopyDone") copyDoneResult, err := pglogrepl.SendStandbyCopyDone(ctx, conn) if err != nil { @@ -621,25 +600,3 @@ loop: return copyDoneResult, nil } - -func (s *Process) sendFeedback(ctx context.Context, conn *pgconn.PgConn, buffer *buffer.Data) { - contextLogger := log.FromContext(ctx) - - err := pglogrepl.SendStandbyStatusUpdate( - ctx, - conn, - pglogrepl.StandbyStatusUpdate{ - WALWritePosition: pglogrepl.LSN(buffer.WriteLSN()), - WALFlushPosition: pglogrepl.LSN(buffer.FlushLSN()), - WALApplyPosition: pglogrepl.LSN(buffer.FlushLSN()), - }, - ) - if err != nil { - contextLogger.Error(err, "Failed to send standby status update, skipping") - } else { - contextLogger.Debug( - "Sent Standby status message", - "write_lsn", types.Int64ToLSN(buffer.WriteLSN()), - "flush_lsn", types.Int64ToLSN(buffer.FlushLSN())) - } -} diff --git a/core/internal/grpc/klio_wal.pb.go b/core/internal/grpc/klio_wal.pb.go index 207d602c..314ca692 100644 --- a/core/internal/grpc/klio_wal.pb.go +++ b/core/internal/grpc/klio_wal.pb.go @@ -47,6 +47,7 @@ type PutRequest struct { WalName string `protobuf:"bytes,2,opt,name=wal_name,json=walName,proto3" json:"wal_name,omitempty"` WalBlock []byte `protobuf:"bytes,3,opt,name=wal_block,json=walBlock,proto3" json:"wal_block,omitempty"` SegmentSize uint64 `protobuf:"varint,4,opt,name=segment_size,json=segmentSize,proto3" json:"segment_size,omitempty"` + WalStartLsn uint64 `protobuf:"varint,8,opt,name=wal_start_lsn,json=walStartLsn,proto3" json:"wal_start_lsn,omitempty"` SendToTier2 bool `protobuf:"varint,7,opt,name=send_to_tier2,json=sendToTier2,proto3" json:"send_to_tier2,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -110,6 +111,13 @@ func (x *PutRequest) GetSegmentSize() uint64 { return 0 } +func (x *PutRequest) GetWalStartLsn() uint64 { + if x != nil { + return x.WalStartLsn + } + return 0 +} + func (x *PutRequest) GetSendToTier2() bool { if x != nil { return x.SendToTier2 @@ -119,7 +127,8 @@ func (x *PutRequest) GetSendToTier2() bool { type PutResult struct { state protoimpl.MessageState `protogen:"open.v1"` - WrittenSize uint64 `protobuf:"varint,1,opt,name=written_size,json=writtenSize,proto3" json:"written_size,omitempty"` + WriteLsn uint64 `protobuf:"varint,3,opt,name=write_lsn,json=writeLsn,proto3" json:"write_lsn,omitempty"` + FlushLsn uint64 `protobuf:"varint,2,opt,name=flush_lsn,json=flushLsn,proto3" json:"flush_lsn,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -154,9 +163,16 @@ func (*PutResult) Descriptor() ([]byte, []int) { return file_proto_klio_wal_proto_rawDescGZIP(), []int{1} } -func (x *PutResult) GetWrittenSize() uint64 { +func (x *PutResult) GetWriteLsn() uint64 { + if x != nil { + return x.WriteLsn + } + return 0 +} + +func (x *PutResult) GetFlushLsn() uint64 { if x != nil { - return x.WrittenSize + return x.FlushLsn } return 0 } @@ -880,16 +896,18 @@ var File_proto_klio_wal_proto protoreflect.FileDescriptor const file_proto_klio_wal_proto_rawDesc = "" + "\n" + - "\x14proto/klio_wal.proto\x12\vklio.wal.v1\x1a\x1fgoogle/protobuf/timestamp.proto\"\xcd\x01\n" + + "\x14proto/klio_wal.proto\x12\vklio.wal.v1\x1a\x1fgoogle/protobuf/timestamp.proto\"\xf1\x01\n" + "\n" + "PutRequest\x12!\n" + "\fcluster_name\x18\x01 \x01(\tR\vclusterName\x12\x19\n" + "\bwal_name\x18\x02 \x01(\tR\awalName\x12\x1b\n" + "\twal_block\x18\x03 \x01(\fR\bwalBlock\x12!\n" + "\fsegment_size\x18\x04 \x01(\x04R\vsegmentSize\x12\"\n" + - "\rsend_to_tier2\x18\a \x01(\bR\vsendToTier2J\x04\b\x05\x10\x06J\x04\b\x06\x10\aR\btrace_idR\aspan_id\".\n" + - "\tPutResult\x12!\n" + - "\fwritten_size\x18\x01 \x01(\x04R\vwrittenSize\"7\n" + + "\rwal_start_lsn\x18\b \x01(\x04R\vwalStartLsn\x12\"\n" + + "\rsend_to_tier2\x18\a \x01(\bR\vsendToTier2J\x04\b\x05\x10\x06J\x04\b\x06\x10\aR\btrace_idR\aspan_id\"Y\n" + + "\tPutResult\x12\x1b\n" + + "\twrite_lsn\x18\x03 \x01(\x04R\bwriteLsn\x12\x1b\n" + + "\tflush_lsn\x18\x02 \x01(\x04R\bflushLsnJ\x04\b\x01\x10\x02R\fwritten_size\"7\n" + "\x12GetMetadataRequest\x12!\n" + "\fcluster_name\x18\x01 \x01(\tR\vclusterName\"J\n" + "\n" + @@ -934,9 +952,9 @@ const file_proto_klio_wal_proto_rawDesc = "" + "\x16tier2_retention_policy\x18\t \x01(\tR\x14tier2RetentionPolicy\"f\n" + "\x11CloseBackupResult\x12%\n" + "\x0etier2_schedule\x18\x01 \x01(\bR\rtier2Schedule\x12*\n" + - "\x11missing_wal_files\x18\x02 \x03(\tR\x0fmissingWalFiles2\xd8\x03\n" + - "\x03WAL\x12:\n" + - "\x03Put\x12\x17.klio.wal.v1.PutRequest\x1a\x16.klio.wal.v1.PutResult\"\x00(\x01\x12:\n" + + "\x11missing_wal_files\x18\x02 \x03(\tR\x0fmissingWalFiles2\xda\x03\n" + + "\x03WAL\x12<\n" + + "\x03Put\x12\x17.klio.wal.v1.PutRequest\x1a\x16.klio.wal.v1.PutResult\"\x00(\x010\x01\x12:\n" + "\x03Get\x12\x17.klio.wal.v1.GetRequest\x1a\x16.klio.wal.v1.GetResult\"\x000\x01\x12N\n" + "\vGetMetadata\x12\x1f.klio.wal.v1.GetMetadataRequest\x1a\x1c.klio.wal.v1.ClusterMetadata\"\x00\x12\\\n" + "\x0fRequestWALStart\x12#.klio.wal.v1.RequestWALStartRequest\x1a\".klio.wal.v1.RequestWALStartResult\"\x00\x12Y\n" + diff --git a/core/internal/grpc/klio_wal_grpc.pb.go b/core/internal/grpc/klio_wal_grpc.pb.go index c78a82b5..f8d46c9b 100644 --- a/core/internal/grpc/klio_wal_grpc.pb.go +++ b/core/internal/grpc/klio_wal_grpc.pb.go @@ -50,7 +50,7 @@ const ( // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type WALClient interface { - Put(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutRequest, PutResult], error) + Put(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[PutRequest, PutResult], error) Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetResult], error) GetMetadata(ctx context.Context, in *GetMetadataRequest, opts ...grpc.CallOption) (*ClusterMetadata, error) RequestWALStart(ctx context.Context, in *RequestWALStartRequest, opts ...grpc.CallOption) (*RequestWALStartResult, error) @@ -66,7 +66,7 @@ func NewWALClient(cc grpc.ClientConnInterface) WALClient { return &wALClient{cc} } -func (c *wALClient) Put(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutRequest, PutResult], error) { +func (c *wALClient) Put(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[PutRequest, PutResult], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &WAL_ServiceDesc.Streams[0], WAL_Put_FullMethodName, cOpts...) if err != nil { @@ -77,7 +77,7 @@ func (c *wALClient) Put(ctx context.Context, opts ...grpc.CallOption) (grpc.Clie } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type WAL_PutClient = grpc.ClientStreamingClient[PutRequest, PutResult] +type WAL_PutClient = grpc.BidiStreamingClient[PutRequest, PutResult] func (c *wALClient) Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetResult], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) @@ -142,7 +142,7 @@ func (c *wALClient) CloseBackup(ctx context.Context, in *CloseBackupRequest, opt // All implementations must embed UnimplementedWALServer // for forward compatibility. type WALServer interface { - Put(grpc.ClientStreamingServer[PutRequest, PutResult]) error + Put(grpc.BidiStreamingServer[PutRequest, PutResult]) error Get(*GetRequest, grpc.ServerStreamingServer[GetResult]) error GetMetadata(context.Context, *GetMetadataRequest) (*ClusterMetadata, error) RequestWALStart(context.Context, *RequestWALStartRequest) (*RequestWALStartResult, error) @@ -158,7 +158,7 @@ type WALServer interface { // pointer dereference when methods are called. type UnimplementedWALServer struct{} -func (UnimplementedWALServer) Put(grpc.ClientStreamingServer[PutRequest, PutResult]) error { +func (UnimplementedWALServer) Put(grpc.BidiStreamingServer[PutRequest, PutResult]) error { return status.Error(codes.Unimplemented, "method Put not implemented") } func (UnimplementedWALServer) Get(*GetRequest, grpc.ServerStreamingServer[GetResult]) error { @@ -202,7 +202,7 @@ func _WAL_Put_Handler(srv interface{}, stream grpc.ServerStream) error { } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type WAL_PutServer = grpc.ClientStreamingServer[PutRequest, PutResult] +type WAL_PutServer = grpc.BidiStreamingServer[PutRequest, PutResult] func _WAL_Get_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(GetRequest) @@ -315,6 +315,7 @@ var WAL_ServiceDesc = grpc.ServiceDesc{ { StreamName: "Put", Handler: _WAL_Put_Handler, + ServerStreams: true, ClientStreams: true, }, { diff --git a/core/internal/repository/writer.go b/core/internal/repository/writer.go index 9c9411e5..01c8ec86 100644 --- a/core/internal/repository/writer.go +++ b/core/internal/repository/writer.go @@ -57,6 +57,9 @@ type WriterOptions struct { // SegmentSize is the length, in bytes, of the WAL segment. SegmentSize uint64 + // The LSN corresponding to the start of the WAL segment. + WALStartLSN uint64 + // Metrics collects per-write metrics for this Writer. Metrics *Metrics diff --git a/core/internal/server/walserver/block_receiver.go b/core/internal/server/walserver/block_receiver.go new file mode 100644 index 00000000..4543c25e --- /dev/null +++ b/core/internal/server/walserver/block_receiver.go @@ -0,0 +1,137 @@ +package walserver + +import ( + "context" + "io" + + "github.com/cloudnative-pg/klio/core/internal/grpc" +) + +// blockReceiverBufferSize bounds how many blocks can be queued ahead of the +// caller before the background goroutine blocks trying to enqueue another +// one. +const blockReceiverBufferSize = 500 + +// blockReceiver decouples receiving WAL blocks from the client's send +// cadence. A background goroutine blocks on req.Recv in a tight loop and +// publishes every block on a channel, so Drain can collect however many +// blocks are already queued on the wire without an artificial wait - +// batching them into a single write+flush instead of one fsync per block. +// +// req.Recv is not safe for concurrent use, so the background goroutine owns +// it exclusively: nothing else may call req.Recv for the lifetime of this +// blockReceiver. +type blockReceiver struct { + ch chan *grpc.PutRequest + cancel context.CancelFunc + done chan struct{} + + // err is set by the background goroutine, once, before it closes ch. + err error +} + +// newBlockReceiverFromKlioClient starts the background receive loop for req. +func newBlockReceiverFromKlioClient(ctx context.Context, req grpc.WAL_PutServer) *blockReceiver { + ctx, cancel := context.WithCancel(ctx) + r := &blockReceiver{ + ch: make(chan *grpc.PutRequest, blockReceiverBufferSize), + cancel: cancel, + done: make(chan struct{}), + } + + go r.run(ctx, req) + + return r +} + +// Drain returns every block currently queued, blocking only if none are +// available yet. Once the stream ends, a non-nil error is only ever +// returned together with a nil batch - io.EOF marks a clean end of stream, +// any other error a failed read. Any blocks still buffered when the stream +// ends are returned first, with a nil error; the closed channel stays +// closed, so the very next call sees it immediately and returns the error. +func (r *blockReceiver) Drain(ctx context.Context) ([]*grpc.PutRequest, error) { + var batch []*grpc.PutRequest + + select { + case request, ok := <-r.ch: + if !ok { + return nil, r.closedErr() + } + batch = append(batch, request) + case <-ctx.Done(): + return nil, ctx.Err() + } + + for { + select { + case request, ok := <-r.ch: + if !ok { + return batch, nil + } + batch = append(batch, request) + default: + return batch, nil + } + } +} + +// Cancel signals the background goroutine to stop, without waiting for it +// to actually exit. Safe to call even while req.Recv is blocked waiting for +// more data from the client - see the "early-exit" note on Stop. +func (r *blockReceiver) Cancel() { + r.cancel() +} + +// Stop cancels the background receive loop and waits for it to exit. It is +// only safe to call from the normal end-of-stream path, i.e. once Drain has +// already returned a non-nil error - at that point req.Recv has already +// returned and the goroutine is exiting or already gone, so the wait is +// immediate. +// +// Do not call Stop from an early-exit path (e.g. a write error while more +// blocks may still be in flight): req.Recv only unblocks on data, client +// half-close, or the RPC's own context ending, and this ctx (a child of +// h.req.Context(), the RPC's own context) has no way to make that happen +// synchronously - so waiting here could block the handler's return +// indefinitely. Use Cancel instead: it only signals, so the handler returns +// right away, and the eventual cleanup is guaranteed anyway - once Put +// returns, the grpc-go runtime cancels h.req.Context() itself as part of +// ending the RPC, which is what actually unblocks the pending req.Recv and +// lets the abandoned goroutine exit on its own. Nothing else touches this +// blockReceiver's channel or req.Recv in the meantime, so a detached +// goroutine is safe. +func (r *blockReceiver) Stop() { + r.cancel() + <-r.done +} + +func (r *blockReceiver) run(ctx context.Context, req grpc.WAL_PutServer) { + defer close(r.ch) + defer close(r.done) + + for { + request, err := req.Recv() + if err != nil { + r.err = err + + return + } + + select { + case r.ch <- request: + case <-ctx.Done(): + r.err = ctx.Err() + + return + } + } +} + +func (r *blockReceiver) closedErr() error { + if r.err != nil { + return r.err + } + + return io.EOF +} diff --git a/core/internal/server/walserver/feedback_sender.go b/core/internal/server/walserver/feedback_sender.go new file mode 100644 index 00000000..c6bd1050 --- /dev/null +++ b/core/internal/server/walserver/feedback_sender.go @@ -0,0 +1,102 @@ +package walserver + +import ( + "context" + "sync/atomic" + + "github.com/cloudnative-pg/machinery/pkg/log" + + "github.com/cloudnative-pg/klio/core/internal/grpc" +) + +// feedbackSenderToKlioClient decouples sending WAL Put progress acks to the client from +// the caller's write/flush cadence. SetWrittenSize never blocks on network +// I/O: it just records the latest known size, since writtenSize is a +// monotonically increasing cumulative total and therefore always +// supersedes an older, unsent value. A background goroutine wakes up on +// each update and sends that latest size, so a batch of blocks written and +// flushed together produces a single ack instead of one per block. +// +// pending is never reset after being sent: it always holds "the latest +// known size", so a wake-up with nothing new to report (e.g. Stop firing +// right after a send already carried the latest size) just resends that +// same value - redundant, but harmless, since only the latest value is +// ever meaningful to the client. 0 is the one value treated as "nothing to +// report yet" and is never sent. +// +// req.Send is not safe for concurrent use: Stop must be called, and must +// return, before the caller uses req.Send again directly - e.g. before +// finalize sends the closing PutResult. +type feedbackSenderToKlioClient struct { + writeLSN atomic.Uint64 + flushLSN atomic.Uint64 + + wake chan struct{} + cancel context.CancelFunc + done chan struct{} +} + +// newFeedbackSenderToKlioClient starts the background send loop for req. +func newFeedbackSenderToKlioClient(ctx context.Context, req grpc.WAL_PutServer) *feedbackSenderToKlioClient { + ctx, cancel := context.WithCancel(ctx) + s := &feedbackSenderToKlioClient{ + wake: make(chan struct{}, 1), + cancel: cancel, + done: make(chan struct{}), + } + + go s.run(ctx, req) + + return s +} + +// SetWriteLSN queues write_lsn to be acked. +func (s *feedbackSenderToKlioClient) SetFeedback(writeLSN uint64, flushLSN uint64) { + if writeLSN == 0 || flushLSN == 0 { + return + } + + s.writeLSN.Store(writeLSN) + s.flushLSN.Store(flushLSN) + + select { + case s.wake <- struct{}{}: + default: + } +} + +// Stop cancels the background goroutine, which flushes any size queued via +// SetWrittenSize but not yet sent before it exits, then waits for it to do +// so. Safe to call more than once. +func (s *feedbackSenderToKlioClient) Stop() { + s.cancel() + <-s.done +} + +func (s *feedbackSenderToKlioClient) run(ctx context.Context, req grpc.WAL_PutServer) { + defer close(s.done) + + for { + select { + case <-s.wake: + s.sendLatest(ctx, req) + case <-ctx.Done(): + s.sendLatest(ctx, req) + return + } + } +} + +func (s *feedbackSenderToKlioClient) sendLatest(ctx context.Context, req grpc.WAL_PutServer) { + if err := req.Send( + &grpc.PutResult{ + WriteLsn: s.writeLSN.Load(), + FlushLsn: s.flushLSN.Load(), + }, + ); err != nil { + log.FromContext(ctx).Error( + err, + "Error while sending feedback to the client, skipping", + ) + } +} diff --git a/core/internal/server/walserver/upload.go b/core/internal/server/walserver/upload.go index bb37dac5..e3d48fe4 100644 --- a/core/internal/server/walserver/upload.go +++ b/core/internal/server/walserver/upload.go @@ -25,7 +25,6 @@ import ( "fmt" "io" "path" - "strconv" "time" "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres" @@ -57,11 +56,33 @@ func (e *incoherentRequestError) Error() string { ) } +// setOrVerify records field's first value and rejects any later value that +// disagrees, so a stream can't silently switch identity (cluster, WAL name, +// segment size, start LSN) partway through. +func setOrVerify[T comparable](field *T, incoming T, fieldName string) error { + var zero T + if *field == zero { + *field = incoming + return nil + } + + if *field != incoming { + return &incoherentRequestError{ + involvedField: fieldName, + expectedValue: fmt.Sprint(*field), + foundValue: fmt.Sprint(incoming), + } + } + + return nil +} + type walUploadBlockMetadata struct { clusterName string walFileName string segmentSize uint64 sendToTier2 bool + walStartLSN uint64 } func (m *walUploadBlockMetadata) handleRequest(request *grpc.PutRequest) error { @@ -75,34 +96,17 @@ func (m *walUploadBlockMetadata) handleRequest(request *grpc.PutRequest) error { return errEmptySegmentSize } - if m.clusterName == "" { - m.clusterName = request.GetClusterName() - } else if m.clusterName != request.GetClusterName() { - return &incoherentRequestError{ - involvedField: "cluster name", - expectedValue: m.clusterName, - foundValue: request.GetClusterName(), - } + if err := setOrVerify(&m.clusterName, request.GetClusterName(), "cluster name"); err != nil { + return err } - - if m.walFileName == "" { - m.walFileName = request.GetWalName() - } else if m.walFileName != request.GetWalName() { - return &incoherentRequestError{ - involvedField: "wal name", - expectedValue: m.walFileName, - foundValue: request.GetWalName(), - } + if err := setOrVerify(&m.walFileName, request.GetWalName(), "wal name"); err != nil { + return err } - - if m.segmentSize == 0 { - m.segmentSize = request.GetSegmentSize() - } else if m.segmentSize != request.GetSegmentSize() { - return &incoherentRequestError{ - involvedField: "wal segment size", - expectedValue: strconv.FormatUint(m.segmentSize, 10), - foundValue: strconv.FormatUint(request.GetSegmentSize(), 10), - } + if err := setOrVerify(&m.segmentSize, request.GetSegmentSize(), "wal segment size"); err != nil { + return err + } + if err := setOrVerify(&m.walStartLSN, request.GetWalStartLsn(), "wal start lsn"); err != nil { + return err } m.sendToTier2 = request.GetSendToTier2() @@ -117,9 +121,11 @@ type putHandler struct { logger log.Logger startTime time.Time - blockMeta walUploadBlockMetadata - walBuffer *repository.Writer - writtenSize uint64 + blockMeta walUploadBlockMetadata + walBuffer *repository.Writer + + writtenBytes uint64 + flushedBytes uint64 // spanEnriched records whether the cluster and WAL name have already been // attached to the RPC span, so we set them only once per Put call. @@ -142,61 +148,118 @@ func (w *Implementation) Put(req grpc.WAL_PutServer) error { return h.run(req.Context()) } -// run consumes the stream of WAL blocks and finalizes the upload. A receive -// error (including io.EOF) stops the loop and the upload is finalized with -// whatever has been written so far. +// run consumes the stream of WAL blocks and finalizes the upload. +// +// Blocks are received on a background goroutine (blockReceiver) and drained +// in batches, so a burst of blocks already queued on the wire is written +// and flushed together - one fsync per batch instead of one per block. +// Progress acks are similarly coalesced by a background goroutine +// (feedbackSenderToKlioClient), so a fast batch of writes doesn't turn into +// a matching burst of synchronous sends back to the client. +// +// A receive error (including io.EOF) stops the loop and the upload is +// finalized with whatever has been written so far - this matches the +// previous behavior of receiveBlock, which never distinguished EOF from +// other read errors for control flow purposes. +// +// Neither receiver nor feedback is stopped via defer: finalize sends the +// closing PutResult directly on h.req, and req.Send is not safe to call +// concurrently with feedback's own background sends, so feedback must be +// stopped before every return. receiver additionally can't always be +// stopped the same way - see the two exit branches below. func (h *putHandler) run(ctx context.Context) error { + receiver := newBlockReceiverFromKlioClient(ctx, h.req) + feedback := newFeedbackSenderToKlioClient(ctx, h.req) + for { - request, err := h.receiveBlock() - if err != nil { - break + batch, recvErr := receiver.Drain(ctx) + if recvErr != nil { + // Drain only ever returns an error together with an empty + // batch (see blockReceiver.Drain), at which point req.Recv has + // already returned: Stop can only be a fast join here, never a + // hang. + receiver.Stop() + feedback.Stop() + + if !errors.Is(recvErr, io.EOF) { + h.logger.Warning( + "Error while reading WAL block", + "clusterName", h.blockMeta.clusterName, + "walName", h.blockMeta.walFileName, + "err", recvErr, + ) + } + + return h.finalize(ctx) } - if err := h.processBlock(ctx, request); err != nil { + if err := h.processBatch(ctx, batch, feedback); err != nil { + // req.Recv may still be blocked waiting on the client here. + // Signal only, don't join: returning err below is what + // eventually unblocks it, once grpc-go cancels h.req.Context() + // as part of ending the RPC - see blockReceiver.Stop. + receiver.Cancel() + feedback.Stop() + return err } } - - return h.finalize(ctx) } -// receiveBlock reads the next WAL block from the stream. It returns an error -// (including io.EOF) once the stream is exhausted or fails. Per-block send -// latency is measured on the sender side (the client's `send` stage), not here: -// this call's duration is dominated by waiting for the client to produce the -// next block, which is gated by PostgreSQL's WAL generation rate. -func (h *putHandler) receiveBlock() (*grpc.PutRequest, error) { - request, err := h.req.Recv() - if err != nil { - if !errors.Is(err, io.EOF) { - h.logger.Warning( - "Error while reading WAL block", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), - "err", err, - ) +// processBatch validates every block in a batch, then writes their payloads +// as a single concatenated write, followed by one flush and one feedback +// update for the whole batch. Concatenating first matters as much as the +// single flush: the WAL writer wraps (compresses/encrypts) and +// length-prefixes each write independently, so writing block-by-block would +// still turn a burst of small blocks into a burst of tiny wrapped chunks in +// the WAL file, even though they'd all share one fsync. +func (h *putHandler) processBatch( + ctx context.Context, + batch []*grpc.PutRequest, + feedback *feedbackSenderToKlioClient, +) error { + var payload []byte + + for _, request := range batch { + if err := h.validateBlock(ctx, request); err != nil { + return err } - return nil, err + payload = append(payload, request.GetWalBlock()...) } - return request, nil -} + if err := h.writeBlock(ctx, payload); err != nil { + return err + } -// processBlock validates a received block, writes it to the WAL buffer and -// updates the latest-written metrics. -func (h *putHandler) processBlock(ctx context.Context, request *grpc.PutRequest) error { - if err := h.validateRequest(request); err != nil { + h.writtenBytes += uint64(len(payload)) + feedback.SetFeedback(h.writtenBytes+h.blockMeta.walStartLSN, h.flushedBytes+h.blockMeta.walStartLSN) + + if err := h.flushBuffer(ctx); err != nil { return err } - h.logger.Debug( - "Received WAL block", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), - "blockLen", len(request.GetWalBlock()), + h.flushedBytes += uint64(len(payload)) + feedback.SetFeedback(h.writtenBytes+h.blockMeta.walStartLSN, h.flushedBytes+h.blockMeta.walStartLSN) + + h.impl.metrics.LatestWrittenLSN.Record( + ctx, + int64(h.flushedBytes+h.blockMeta.walStartLSN), //nolint:gosec + metric.WithAttributeSet( + h.impl.metrics.AttributeSet(opentelemetry.AttributeKeyClusterName.Of(h.blockMeta.clusterName)), + ), ) + return nil +} + +// validateBlock validates a single received block and updates the shared +// metadata. It does not write the block's payload - see processBatch. +func (h *putHandler) validateBlock(ctx context.Context, request *grpc.PutRequest) error { + if err := h.validateRequest(request); err != nil { + return err + } + if err := h.blockMeta.handleRequest(request); err != nil { h.logger.Error( err, @@ -211,18 +274,7 @@ func (h *putHandler) processBlock(ctx context.Context, request *grpc.PutRequest) h.enrichSpan(ctx) h.recordLatestWrittenTimeline(ctx) - if err := h.openWriter(request); err != nil { - return err - } - - if err := h.writeBlock(ctx, request); err != nil { - return err - } - - h.writtenSize += uint64(len(request.GetWalBlock())) - h.recordLatestWrittenLSN(ctx) - - return nil + return h.openWriter(request) } // validateRequest checks the cluster name and WAL name of a received block. @@ -244,6 +296,13 @@ func (h *putHandler) validateRequest(request *grpc.PutRequest) error { return nil } +// putWriteBufferSize is the write buffer size for the WAL writer opened by +// a Put call. Batches of blocks already share a single WriteBlock call (see +// processBatch), so this mainly bounds the buffer for the rare batch that +// exceeds it, coalescing the underlying writes rather than trickling them +// out block-by-block. +const putWriteBufferSize = 4 * 1024 * 1024 + // openWriter lazily creates the WAL buffer writer on the first received block. func (h *putHandler) openWriter(request *grpc.PutRequest) error { if h.walBuffer != nil { @@ -256,6 +315,8 @@ func (h *putHandler) openWriter(request *grpc.PutRequest) error { WALName: h.blockMeta.walFileName, SegmentSize: h.blockMeta.segmentSize, Metrics: h.impl.metrics, + BufferSize: putWriteBufferSize, + WALStartLSN: h.blockMeta.walStartLSN, }, ) if err != nil { @@ -274,19 +335,26 @@ func (h *putHandler) openWriter(request *grpc.PutRequest) error { return nil } -// writeBlock writes and flushes a single WAL block to the buffer. -func (h *putHandler) writeBlock(ctx context.Context, request *grpc.PutRequest) error { - if err := h.walBuffer.WriteBlock(ctx, request.GetWalBlock()); err != nil { +// writeBlock writes a batch's concatenated payload to the buffer, without +// flushing it. +func (h *putHandler) writeBlock(ctx context.Context, data []byte) error { + if err := h.walBuffer.WriteBlock(ctx, data); err != nil { h.logger.Error( err, "Error while writing WAL data", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), + "clusterName", h.blockMeta.clusterName, + "walName", h.blockMeta.walFileName, ) return status.Errorf(grpccodes.Internal, "error while writing WAL: %v", err.Error()) } + return nil +} + +// flushBuffer flushes the WAL buffer, covering every block written to it +// since the last flush. +func (h *putHandler) flushBuffer(ctx context.Context) error { flushStart := time.Now() err := h.walBuffer.Flush() flushDuration := time.Since(flushStart) @@ -294,16 +362,16 @@ func (h *putHandler) writeBlock(ctx context.Context, request *grpc.PutRequest) e h.logger.Error( err, "Error while flushing WAL data", - "clusterName", request.GetClusterName(), - "walName", request.GetWalName(), + "clusterName", h.blockMeta.clusterName, + "walName", h.blockMeta.walFileName, ) - h.impl.metrics.RecordBlockStage(ctx, request.GetClusterName(), opentelemetry.PathPut, + h.impl.metrics.RecordBlockStage(ctx, h.blockMeta.clusterName, opentelemetry.PathPut, opentelemetry.StageFlush, flushDuration, opentelemetry.OutcomeFailure) return status.Errorf(grpccodes.Internal, "error while flushing WAL: %v", err.Error()) } - h.impl.metrics.RecordBlockStage(ctx, request.GetClusterName(), opentelemetry.PathPut, + h.impl.metrics.RecordBlockStage(ctx, h.blockMeta.clusterName, opentelemetry.PathPut, opentelemetry.StageFlush, flushDuration, opentelemetry.OutcomeSuccess) return nil @@ -348,28 +416,6 @@ func (h *putHandler) recordLatestWrittenTimeline(ctx context.Context) { ) } -// recordLatestWrittenLSN updates the latest written LSN metric for real WAL -// segments. Non-segment files are skipped, as for the timeline metric. -func (h *putHandler) recordLatestWrittenLSN(ctx context.Context) { - startPos, err := lsnStartFromWALName(h.blockMeta.walFileName, h.blockMeta.segmentSize) - if errors.Is(err, errNotWALSegment) { - return - } - if err != nil { - h.logger.Error(err, "Could not compute start LSN for latest written LSN metric", - "walName", h.blockMeta.walFileName) - return - } - - h.impl.metrics.LatestWrittenLSN.Record( - ctx, - int64(startPos+h.writtenSize), //nolint:gosec - metric.WithAttributeSet( - h.impl.metrics.AttributeSet(opentelemetry.AttributeKeyClusterName.Of(h.blockMeta.clusterName)), - ), - ) -} - // timelineFromWALName returns the timeline ID of a WAL segment file. It returns // errNotWALSegment for non-segment files (.history, .backup, .partial), which // must not drive the latest_written_* metrics. @@ -417,12 +463,13 @@ func (h *putHandler) finalize(ctx context.Context) error { return err } - if err := h.req.SendAndClose(&grpc.PutResult{ - WrittenSize: h.writtenSize, + if err := h.req.Send(&grpc.PutResult{ + FlushLsn: h.flushedBytes, + WriteLsn: h.writtenBytes, }); err != nil { h.logger.Warning( "Error while sending WAL Put response", - "writtenSize", h.writtenSize, + "flushedLSN", h.flushedBytes, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "err", err, @@ -436,8 +483,9 @@ func (h *putHandler) finalize(ctx context.Context) error { // closeEmpty reports an empty result when no WAL block was ever received. func (h *putHandler) closeEmpty() error { - if err := h.req.SendAndClose(&grpc.PutResult{ - WrittenSize: 0, + if err := h.req.Send(&grpc.PutResult{ + FlushLsn: 0, + WriteLsn: 0, }); err != nil { h.logger.Error(err, "Error while closing empty WAL file") @@ -450,16 +498,16 @@ func (h *putHandler) closeEmpty() error { // closeBuffer closes the WAL buffer, distinguishing between a partial and a // complete WAL segment. func (h *putHandler) closeBuffer(ctx context.Context) error { - if !h.isCompleted() { - return h.closePartial() + if h.isCompleted() { + return h.closeComplete(ctx) } - return h.closeComplete(ctx) + return h.closePartial() } // isCompleted returns true if the WAL segment has been fully received. func (h *putHandler) isCompleted() bool { - return h.writtenSize == h.blockMeta.segmentSize && h.writtenSize != 0 + return h.flushedBytes == h.blockMeta.segmentSize } // closePartial closes a partially received WAL file. @@ -467,7 +515,6 @@ func (h *putHandler) closePartial() error { if err := h.walBuffer.Close(); err != nil { h.logger.Warning( "Error while closing partial WAL file", - "writtenSize", h.writtenSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "err", err, @@ -478,8 +525,6 @@ func (h *putHandler) closePartial() error { h.logger.Info( "Received partial WAL file", - "writtenSize", h.writtenSize, - "segmentSize", h.blockMeta.segmentSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "elapsedTime", time.Since(h.startTime), @@ -493,7 +538,6 @@ func (h *putHandler) closeComplete(ctx context.Context) error { if err := h.walBuffer.CloseMarkDone(); err != nil { h.logger.Warning( "Error while closing completed WAL file", - "writtenSize", h.writtenSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "err", err, @@ -511,8 +555,6 @@ func (h *putHandler) closeComplete(ctx context.Context) error { h.logger.Info( "Received completed WAL file", - "writtenSize", h.writtenSize, - "segmentSize", h.blockMeta.segmentSize, "walFileName", h.blockMeta.walFileName, "clusterName", h.blockMeta.clusterName, "elapsedTime", time.Since(h.startTime), diff --git a/core/internal/server/walserver/upload_test.go b/core/internal/server/walserver/upload_test.go index 97f90d64..ac6f2b56 100644 --- a/core/internal/server/walserver/upload_test.go +++ b/core/internal/server/walserver/upload_test.go @@ -25,37 +25,38 @@ import ( ) // TestIsCompleted verifies that a WAL segment is considered complete only when -// the full segment has been received: a short write (interrupted by a failover) -// or a zero-length write is treated as partial. +// flushedLSN has reached segmentSize: a short write (interrupted by a +// failover) or a zero-length write is treated as partial. func TestIsCompleted(t *testing.T) { tests := []struct { name string - writtenSize uint64 + flushedLSN uint64 + walStartLSN uint64 segmentSize uint64 want bool }{ { name: "fully received segment is complete", - writtenSize: 16, + flushedLSN: 16, segmentSize: 16, want: true, }, { name: "short write is partial", - writtenSize: 8, + flushedLSN: 8, segmentSize: 16, want: false, }, { name: "empty write is partial", - writtenSize: 0, + flushedLSN: 0, segmentSize: 16, want: false, }, { - name: "zero-sized segment is never complete", - writtenSize: 0, - segmentSize: 0, + name: "short write with non-zero start offset is partial", + flushedLSN: 24, + segmentSize: 16, want: false, }, } @@ -63,8 +64,11 @@ func TestIsCompleted(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { h := &putHandler{ - writtenSize: tt.writtenSize, - blockMeta: walUploadBlockMetadata{segmentSize: tt.segmentSize}, + flushedBytes: tt.flushedLSN, + blockMeta: walUploadBlockMetadata{ + segmentSize: tt.segmentSize, + walStartLSN: tt.walStartLSN, + }, } if got := h.isCompleted(); got != tt.want { t.Fatalf("isCompleted() = %v, want %v", got, tt.want) diff --git a/core/internal/wal/feedback.go b/core/internal/wal/feedback.go new file mode 100644 index 00000000..51af2e3d --- /dev/null +++ b/core/internal/wal/feedback.go @@ -0,0 +1,33 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package wal + +// Feedback represents the feedback we get from the Klio server +// when writing a WAL file. +type Feedback struct { + // corresponding to flush_lsn in pg_stat_replication + FlushLSN uint64 + + // corresponding to write_lsn in pg_stat_replication + WriteLSN uint64 + + // corresponding to replay_lsn in pg_stat_replication + ReplayLSN uint64 +} diff --git a/core/internal/walplayer/player.go b/core/internal/walplayer/player.go index 68e90c37..515492e0 100644 --- a/core/internal/walplayer/player.go +++ b/core/internal/walplayer/player.go @@ -247,7 +247,7 @@ func (p *Player) sendWAL(ctx context.Context, c *grpcclient.Connection, fileName return result } - stream, err := c.StoreWALStreaming(ctx, path.Base(fileName), size, false) + stream, err := c.StoreWALStreaming(ctx, path.Base(fileName), size, false, 0, nil) if err != nil { result.Error = fmt.Sprintf("while starting WAL file streaming: %v", err) return result diff --git a/core/pkg/config/client.go b/core/pkg/config/client.go index a6726d59..b8480db8 100644 --- a/core/pkg/config/client.go +++ b/core/pkg/config/client.go @@ -19,8 +19,6 @@ SPDX-License-Identifier: Apache-2.0 package config -import "time" - // Data is the configuration. // // This struct is used to generate a secret in the Kubernetes cluster, so its serialization must be stable. @@ -79,14 +77,6 @@ type SourceConfig struct { // Slot is the name of the replication slot to be used Slot string `json:"slot" mapstructure:"slot"` - // StandbyMessageTimeoutSeconds is the timeout after which the WAL - // receiver will send a status update - StandbyMessageTimeoutSeconds int `json:"standby_message_timeout_seconds" mapstructure:"standby_message_timeout_seconds"` //nolint:lll - - // FlushTimeoutMilliseconds is the timeout in milliseconds after which buffered - // WAL data is automatically flushed to the Klio server - FlushTimeoutMilliseconds int `json:"flush_timeout_ms" mapstructure:"flush_timeout_ms"` - // BufferSize is the maximum size in bytes of the in-memory WAL buffer before // triggering an automatic flush BufferSize int `json:"buffer_size" mapstructure:"buffer_size"` @@ -155,19 +145,5 @@ type WALPrefetchConfig struct { // SetDefaults sets the default values of the configuration. func (s *SourceConfig) SetDefaults() { - s.StandbyMessageTimeoutSeconds = 10 - s.FlushTimeoutMilliseconds = 200 - s.BufferSize = 2 * 1024 * 1024 // 2 MB -} - -// StandbyMessageTimeout returns the stanby message timeout in a -// time.Duration. -func (s *SourceConfig) StandbyMessageTimeout() time.Duration { - return time.Second * time.Duration(s.StandbyMessageTimeoutSeconds) -} - -// FlushTimeout returns the timeout after which the WALs are -// flushed. -func (s *SourceConfig) FlushTimeout() time.Duration { - return time.Millisecond * time.Duration(s.FlushTimeoutMilliseconds) + s.BufferSize = 4 * 1024 * 1024 // 4 MB } diff --git a/core/pkg/config/client_validate.go b/core/pkg/config/client_validate.go index 1690d9e1..82763407 100644 --- a/core/pkg/config/client_validate.go +++ b/core/pkg/config/client_validate.go @@ -58,12 +58,6 @@ func (s *SourceConfig) Validate() error { errs = errors.Join(errs, errors.New( "invalid source config: slot name can only contain lower-case letters, numbers, and underscores")) } - if s.StandbyMessageTimeoutSeconds < 1 { - errs = errors.Join(errs, errors.New("invalid source config: standby_message_timeout_seconds must be at least 1")) - } - if s.FlushTimeoutMilliseconds < 1 { - errs = errors.Join(errs, errors.New("invalid source config: flush_timeout_ms must be at least 1")) - } if s.BufferSize < 1 { errs = errors.Join(errs, errors.New("invalid source config: buffer_size must be at least 1")) } diff --git a/core/pkg/config/client_validate_test.go b/core/pkg/config/client_validate_test.go index 50652a21..061d103d 100644 --- a/core/pkg/config/client_validate_test.go +++ b/core/pkg/config/client_validate_test.go @@ -34,12 +34,10 @@ func TestSourceConfigValidate(t *testing.T) { { name: "Valid config", config: SourceConfig{ - DSN: "postgres://...", - StandardDSN: "postgres://...", - Slot: "my_slot_123", - StandbyMessageTimeoutSeconds: 10, - FlushTimeoutMilliseconds: 100, - BufferSize: 1024, + DSN: "postgres://...", + StandardDSN: "postgres://...", + Slot: "my_slot_123", + BufferSize: 1024, }, wantErr: false, }, @@ -61,41 +59,13 @@ func TestSourceConfigValidate(t *testing.T) { wantErr: true, substr: "slot name can only contain lower-case letters", }, - { - name: "Standby Message Timeout too low", - config: SourceConfig{ - DSN: "valid", - StandardDSN: "valid", - Slot: "valid", - StandbyMessageTimeoutSeconds: 0, - FlushTimeoutMilliseconds: 100, - BufferSize: 1024, - }, - wantErr: true, - substr: "must be at least 1", - }, - { - name: "Flush Timeout too low", - config: SourceConfig{ - DSN: "valid", - StandardDSN: "valid", - Slot: "valid", - StandbyMessageTimeoutSeconds: 100, - FlushTimeoutMilliseconds: 0, - BufferSize: 1024, - }, - wantErr: true, - substr: "must be at least 1", - }, { name: "Buffer Size too low", config: SourceConfig{ - DSN: "valid", - StandardDSN: "valid", - Slot: "valid", - StandbyMessageTimeoutSeconds: 100, - FlushTimeoutMilliseconds: 200, - BufferSize: 0, + DSN: "valid", + StandardDSN: "valid", + Slot: "valid", + BufferSize: 0, }, wantErr: true, substr: "must be at least 1", diff --git a/core/pkg/config/decode_test.go b/core/pkg/config/decode_test.go index d1339b12..cd389d43 100644 --- a/core/pkg/config/decode_test.go +++ b/core/pkg/config/decode_test.go @@ -78,18 +78,14 @@ source: dsn: "postgres://localhost:5432/mydb" standard_dsn: "postgres://localhost:5432/mydb" slot: my_slot - standby_message_timeout_seconds: 15 - flush_timeout_ms: 300 buffer_size: 4096 `, want: Data{ Source: SourceConfig{ - DSN: "postgres://localhost:5432/mydb", - StandardDSN: "postgres://localhost:5432/mydb", - Slot: "my_slot", - StandbyMessageTimeoutSeconds: 15, - FlushTimeoutMilliseconds: 300, - BufferSize: 4096, + DSN: "postgres://localhost:5432/mydb", + StandardDSN: "postgres://localhost:5432/mydb", + Slot: "my_slot", + BufferSize: 4096, }, }, }, diff --git a/core/proto/klio_wal.proto b/core/proto/klio_wal.proto index afbc3c15..0de96277 100644 --- a/core/proto/klio_wal.proto +++ b/core/proto/klio_wal.proto @@ -25,7 +25,7 @@ import "google/protobuf/timestamp.proto"; option go_package = "github.com/cloudnative-pg/klio/core/internal/grpc"; service WAL { - rpc Put(stream PutRequest) returns (PutResult) {} + rpc Put(stream PutRequest) returns (stream PutResult) {} rpc Get(GetRequest) returns (stream GetResult) {} rpc GetMetadata(GetMetadataRequest) returns (ClusterMetadata) {} @@ -40,6 +40,7 @@ message PutRequest { string wal_name = 2; bytes wal_block = 3; uint64 segment_size = 4; + uint64 wal_start_lsn = 8; reserved 5, 6; // was trace_id, span_id reserved "trace_id", "span_id"; @@ -48,7 +49,10 @@ message PutRequest { } message PutResult { - uint64 written_size = 1; + reserved 1; // was written_size + reserved "written_size"; + uint64 write_lsn = 3; + uint64 flush_lsn = 2; } message GetMetadataRequest { diff --git a/documentation/web/docs/developer/_protocol.md b/documentation/web/docs/developer/_protocol.md index 4a1665e2..82c268ac 100644 --- a/documentation/web/docs/developer/_protocol.md +++ b/documentation/web/docs/developer/_protocol.md @@ -404,6 +404,7 @@ file | wal_name | [string](#string) | | | | wal_block | [bytes](#bytes) | | | | segment_size | [uint64](#uint64) | | | +| wal_start_lsn | [uint64](#uint64) | | | | send_to_tier2 | [bool](#bool) | | | @@ -419,7 +420,8 @@ file | Field | Type | Label | Description | | ----- | ---- | ----- | ----------- | -| written_size | [uint64](#uint64) | | | +| write_lsn | [uint64](#uint64) | | | +| flush_lsn | [uint64](#uint64) | | | @@ -539,7 +541,7 @@ feature. | Method Name | Request Type | Response Type | Description | | ----------- | ------------ | ------------- | ------------| -| Put | [PutRequest](#klio-wal-v1-PutRequest) stream | [PutResult](#klio-wal-v1-PutResult) | | +| Put | [PutRequest](#klio-wal-v1-PutRequest) stream | [PutResult](#klio-wal-v1-PutResult) stream | | | Get | [GetRequest](#klio-wal-v1-GetRequest) | [GetResult](#klio-wal-v1-GetResult) stream | | | GetMetadata | [GetMetadataRequest](#klio-wal-v1-GetMetadataRequest) | [ClusterMetadata](#klio-wal-v1-ClusterMetadata) | | | RequestWALStart | [RequestWALStartRequest](#klio-wal-v1-RequestWALStartRequest) | [RequestWALStartResult](#klio-wal-v1-RequestWALStartResult) | | diff --git a/operator/internal/klioconfig/config.go b/operator/internal/klioconfig/config.go index 89b43d0d..6b6f2b76 100644 --- a/operator/internal/klioconfig/config.go +++ b/operator/internal/klioconfig/config.go @@ -104,9 +104,7 @@ func GenerateConfig( StandardDSN: "user=postgres application_name=klio", Slot: "klio", // The following parameters are not used by the plugin, but here with their default for completeness - StandbyMessageTimeoutSeconds: 0, - FlushTimeoutMilliseconds: 0, - BufferSize: 0, + BufferSize: 0, }, Client: config.ClientConfig{ ClusterName: spec.ClusterName, diff --git a/operator/pkg/config/client.go b/operator/pkg/config/client.go index a6726d59..b8480db8 100644 --- a/operator/pkg/config/client.go +++ b/operator/pkg/config/client.go @@ -19,8 +19,6 @@ SPDX-License-Identifier: Apache-2.0 package config -import "time" - // Data is the configuration. // // This struct is used to generate a secret in the Kubernetes cluster, so its serialization must be stable. @@ -79,14 +77,6 @@ type SourceConfig struct { // Slot is the name of the replication slot to be used Slot string `json:"slot" mapstructure:"slot"` - // StandbyMessageTimeoutSeconds is the timeout after which the WAL - // receiver will send a status update - StandbyMessageTimeoutSeconds int `json:"standby_message_timeout_seconds" mapstructure:"standby_message_timeout_seconds"` //nolint:lll - - // FlushTimeoutMilliseconds is the timeout in milliseconds after which buffered - // WAL data is automatically flushed to the Klio server - FlushTimeoutMilliseconds int `json:"flush_timeout_ms" mapstructure:"flush_timeout_ms"` - // BufferSize is the maximum size in bytes of the in-memory WAL buffer before // triggering an automatic flush BufferSize int `json:"buffer_size" mapstructure:"buffer_size"` @@ -155,19 +145,5 @@ type WALPrefetchConfig struct { // SetDefaults sets the default values of the configuration. func (s *SourceConfig) SetDefaults() { - s.StandbyMessageTimeoutSeconds = 10 - s.FlushTimeoutMilliseconds = 200 - s.BufferSize = 2 * 1024 * 1024 // 2 MB -} - -// StandbyMessageTimeout returns the stanby message timeout in a -// time.Duration. -func (s *SourceConfig) StandbyMessageTimeout() time.Duration { - return time.Second * time.Duration(s.StandbyMessageTimeoutSeconds) -} - -// FlushTimeout returns the timeout after which the WALs are -// flushed. -func (s *SourceConfig) FlushTimeout() time.Duration { - return time.Millisecond * time.Duration(s.FlushTimeoutMilliseconds) + s.BufferSize = 4 * 1024 * 1024 // 4 MB }