From 81c571f7054be139aa3a27eddbfd52a64afdd37b Mon Sep 17 00:00:00 2001
From: Gabriele Quaresima <gabriele.quaresima@enterprisedb.com>
Date: Mon, 17 Aug 2026 11:00:22 +0200
Subject: [PATCH 1/2] fix(wal): gate flush feedback on durable server
 acknowledgements

The WAL streaming client reported a flush position to PostgreSQL as soon as
WAL data was handed to the gRPC send buffer, before the Klio server confirmed
it durable. That made it unsafe to use Klio as a synchronous replication
target for zero RPO: PostgreSQL could acknowledge commits, or advance the
replication slot's restart_lsn, for data not yet guaranteed to survive a
server or network failure.

Convert the Put RPC to bidirectional streaming so the server acknowledges each
block once fsynced, and add a require_durable_ack client option (default off)
that advances the flush position only up to durably acknowledged data. The
default preserves the previous optimistic behavior.

Assisted-by: Claude
Signed-off-by: Gabriele Quaresima <gabriele.quaresima@enterprisedb.com>
---
 .../klioclient/grpcclient/connection.go       |  92 ++++++++-
 .../client/klioclient/grpcclient/walclient.go |  29 ++-
 .../klioclient/grpcclient/walstreamer.go      |  14 +-
 core/internal/client/klioclient/interfaces.go |  10 +
 core/internal/client/sendwal/buffer/buffer.go |  82 +++++---
 .../client/sendwal/buffer/buffer_test.go      | 179 ++++++++++++++++++
 core/internal/client/sendwal/buffer/grpc.go   |  10 +
 .../internal/client/sendwal/buffer/handler.go |   5 +
 core/internal/client/sendwal/buffer/memory.go |   6 +
 core/internal/client/sendwal/receiver.go      |   1 +
 core/internal/grpc/klio_wal.pb.go             |  16 +-
 core/internal/grpc/klio_wal_grpc.pb.go        |  13 +-
 core/internal/server/walserver/upload.go      |  45 +++--
 core/pkg/config/client.go                     |   8 +
 core/proto/klio_wal.proto                     |   8 +-
 documentation/web/docs/developer/_protocol.md |   9 +-
 operator/pkg/config/client.go                 |   8 +
 17 files changed, 465 insertions(+), 70 deletions(-)
 create mode 100644 core/internal/client/sendwal/buffer/buffer_test.go

diff --git a/core/internal/client/klioclient/grpcclient/connection.go b/core/internal/client/klioclient/grpcclient/connection.go
index 25064d94..b7604e58 100644
--- a/core/internal/client/klioclient/grpcclient/connection.go
+++ b/core/internal/client/klioclient/grpcclient/connection.go
@@ -23,8 +23,11 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
+	"io"
 	"os"
+	"sync/atomic"
 
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
@@ -42,18 +45,75 @@ type grpcWALStream struct {
 	sentBytes   uint64
 	walName     string
 	sendToTier2 bool
+
+	// syncedBytes is the highest cumulative durable size acknowledged by the
+	// server for this WAL file. It is written by the ack reader goroutine and
+	// read by the WAL processing goroutine, so it must be accessed atomically.
+	syncedBytes atomic.Uint64
+
+	// ackDone is closed when the ack reader goroutine has consumed every
+	// acknowledgement up to the end of the stream. ackErr, set before ackDone
+	// is closed, carries any non-EOF error observed while reading.
+	ackDone chan struct{}
+	ackErr  error
+}
+
+// newGRPCWALStream wraps a freshly opened Put stream and starts the background
+// goroutine that consumes the server durability acknowledgements.
+func newGRPCWALStream(
+	stream klioGRPC.WAL_PutClient,
+	name string,
+	segmentSize uint64,
+	clusterName string,
+	sendToTier2 bool,
+) *grpcWALStream {
+	g := &grpcWALStream{
+		innerStream: stream,
+		segmentSize: segmentSize,
+		clusterName: clusterName,
+		walName:     name,
+		sendToTier2: sendToTier2,
+		ackDone:     make(chan struct{}),
+	}
+
+	go g.readAcks()
+
+	return g
+}
+
+// SyncedOffset implements common.WALStream. It returns the highest durable size
+// acknowledged so far for this WAL file.
+func (g *grpcWALStream) SyncedOffset() (uint64, error) {
+	// If the reader has already terminated with a non-EOF error, the
+	// acknowledgements can no longer be trusted.
+	select {
+	case <-g.ackDone:
+		if g.ackErr != nil {
+			return 0, fmt.Errorf("while receiving WAL acknowledgements: %w", g.ackErr)
+		}
+	default:
+	}
+
+	return g.syncedBytes.Load(), nil
 }
 
-// Close implements common.WALStream.
+// Close implements common.WALStream. It stops sending, waits for every
+// acknowledgement to be drained and verifies that the whole file is durable.
 func (g *grpcWALStream) Close(_ context.Context) error {
-	result, err := g.innerStream.CloseAndRecv()
-	if err != nil {
-		return fmt.Errorf("while flushing WAL file: %w", err)
+	if err := g.innerStream.CloseSend(); err != nil {
+		return fmt.Errorf("while closing the WAL upload stream: %w", err)
+	}
+
+	// Wait for the reader to drain all acknowledgements up to the end of the
+	// stream, so syncedBytes reflects the final durable size.
+	<-g.ackDone
+	if g.ackErr != nil {
+		return fmt.Errorf("while receiving WAL acknowledgements: %w", g.ackErr)
 	}
 
-	if result.GetWrittenSize() != g.sentBytes {
+	if synced := g.syncedBytes.Load(); synced != g.sentBytes {
 		return &IncompleteWALFileError{
-			uploadedSize: result.GetWrittenSize(),
+			uploadedSize: synced,
 			expectedSize: g.sentBytes,
 		}
 	}
@@ -61,6 +121,26 @@ func (g *grpcWALStream) Close(_ context.Context) error {
 	return nil
 }
 
+// readAcks consumes the server acknowledgements until the stream ends, tracking
+// the latest durable size. gRPC allows a single concurrent reader alongside the
+// single writer used by SendBlock.
+func (g *grpcWALStream) readAcks() {
+	defer close(g.ackDone)
+
+	for {
+		result, err := g.innerStream.Recv()
+		if err != nil {
+			if !errors.Is(err, io.EOF) {
+				g.ackErr = err
+			}
+
+			return
+		}
+
+		g.syncedBytes.Store(result.GetWrittenSize())
+	}
+}
+
 // Connection represents a connection to a Klio server.
 type Connection struct {
 	klioGRPC.WALClient
diff --git a/core/internal/client/klioclient/grpcclient/walclient.go b/core/internal/client/klioclient/grpcclient/walclient.go
index 388280fc..5d317ce9 100644
--- a/core/internal/client/klioclient/grpcclient/walclient.go
+++ b/core/internal/client/klioclient/grpcclient/walclient.go
@@ -61,14 +61,18 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte,
 		}
 	}
 
-	result, err := stream.CloseAndRecv()
+	if err := stream.CloseSend(); err != nil {
+		return fmt.Errorf("while closing the WAL upload stream: %w", err)
+	}
+
+	writtenSize, err := drainPutAcks(stream)
 	if err != nil {
-		return fmt.Errorf("while flushing WAL file: %w", err)
+		return err
 	}
 
-	if result.GetWrittenSize() != uint64(len(content)) {
+	if writtenSize != uint64(len(content)) {
 		return &IncompleteWALFileError{
-			uploadedSize: result.GetWrittenSize(),
+			uploadedSize: writtenSize,
 			expectedSize: uint64(len(content)),
 		}
 	}
@@ -76,6 +80,23 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte,
 	return nil
 }
 
+// drainPutAcks consumes the durability acknowledgements of a Put stream until
+// it ends, returning the last (cumulative) durable size reported by the server.
+func drainPutAcks(stream klioGRPC.WAL_PutClient) (uint64, error) {
+	var writtenSize uint64
+	for {
+		result, err := stream.Recv()
+		if errors.Is(err, io.EOF) {
+			return writtenSize, nil
+		}
+		if err != nil {
+			return 0, fmt.Errorf("while flushing WAL file: %w", err)
+		}
+
+		writtenSize = result.GetWrittenSize()
+	}
+}
+
 // 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/walstreamer.go
index 327afb23..0583751c 100644
--- a/core/internal/client/klioclient/grpcclient/walstreamer.go
+++ b/core/internal/client/klioclient/grpcclient/walstreamer.go
@@ -46,13 +46,13 @@ func (c *Connection) StoreWALStreaming(
 		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
+	return klioclient.NewWALUploader(newGRPCWALStream(
+		stream,
+		name,
+		segmentSize,
+		c.clientConfig.ClusterName,
+		sendToTier2,
+	)), nil
 }
 
 // GetWALStreaming get a WAL from a remote connection.
diff --git a/core/internal/client/klioclient/interfaces.go b/core/internal/client/klioclient/interfaces.go
index 6c4642f4..d0f17881 100644
--- a/core/internal/client/klioclient/interfaces.go
+++ b/core/internal/client/klioclient/interfaces.go
@@ -103,6 +103,10 @@ type WALUploaderImpl interface {
 	// SendBlock sends a WAL Block
 	SendBlock(ctx context.Context, block []byte) error
 
+	// SyncedOffset returns the number of bytes the server has acknowledged as
+	// durably persisted for the current WAL file.
+	SyncedOffset() (uint64, error)
+
 	// Close closes the WAL streaming session
 	Close(ctx context.Context) error
 }
@@ -124,6 +128,12 @@ func (u *WALUploader) SendBlock(ctx context.Context, block []byte) error {
 	return u.impl.SendBlock(ctx, block) //nolint:wrapcheck
 }
 
+// SyncedOffset returns the number of bytes the server has acknowledged as
+// durably persisted for the current WAL file.
+func (u *WALUploader) SyncedOffset() (uint64, error) {
+	return u.impl.SyncedOffset() //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..2b7c9971 100644
--- a/core/internal/client/sendwal/buffer/buffer.go
+++ b/core/internal/client/sendwal/buffer/buffer.go
@@ -40,19 +40,27 @@ type Data struct {
 
 	handler Handler
 
-	writeLSN   uint64
-	flushLSN   uint64
-	buffer     *bytes.Buffer
-	bufferSize int
+	writeLSN     uint64
+	flushLSN     uint64
+	fileStartLSN uint64
+	buffer       *bytes.Buffer
+	bufferSize   int
+
+	// requireDurableAck gates flushLSN advancement on the destination
+	// acknowledging that the data has been durably persisted, instead of
+	// advancing it as soon as the data has been handed to the send buffer.
+	requireDurableAck bool
 }
 
-// New creates a new WAL buffer.
-func New(tli int, walSegmentSize uint64, handler Handler, bufferSize int) *Data {
+// New creates a new WAL buffer. When requireDurableAck is true, the flush LSN
+// only advances up to data the handler reports as durably persisted.
+func New(tli int, walSegmentSize uint64, handler Handler, bufferSize int, requireDurableAck bool) *Data {
 	result := &Data{
-		segmentSize: walSegmentSize,
-		tli:         tli,
-		handler:     handler,
-		bufferSize:  bufferSize,
+		segmentSize:       walSegmentSize,
+		tli:               tli,
+		handler:           handler,
+		bufferSize:        bufferSize,
+		requireDurableAck: requireDurableAck,
 	}
 
 	result.buffer = result.newBuffer()
@@ -167,6 +175,7 @@ func (wal *Data) openWALPos(ctx context.Context, blockpos uint64) error {
 
 	wal.writeLSN = blockpos
 	wal.flushLSN = blockpos
+	wal.fileStartLSN = blockpos
 
 	return nil
 }
@@ -188,26 +197,50 @@ func (wal *Data) writeToWALFile(ctx context.Context, data []byte) error {
 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.handler == nil || !wal.handler.HasWALFileOpened() {
 		return nil
 	}
 
-	contextLogger.Debug("Writing block",
-		"blockpos", types.Int64ToLSN(wal.writeLSN), "blocksize", wal.buffer.Len())
-	_, err := wal.handler.Write(ctx, wal.buffer.Bytes())
-	if err != nil {
-		return fmt.Errorf("while writing to WAL handler: %w", err)
+	if wal.buffer.Len() > 0 {
+		contextLogger.Debug("Writing block",
+			"blockpos", types.Int64ToLSN(wal.writeLSN), "blocksize", wal.buffer.Len())
+		_, err := wal.handler.Write(ctx, wal.buffer.Bytes())
+		if err != nil {
+			return fmt.Errorf("while writing to WAL handler: %w", err)
+		}
+
+		// Clear content but keeps the slice capacity
+		wal.buffer.Reset()
+
+		// Prevent memory bloat in long-running processes.
+		if wal.buffer.Cap() > wal.bufferSize*maximumBufferSizeFactor {
+			wal.buffer = wal.newBuffer()
+		}
 	}
 
-	// Clear content but keeps the slice capacity
-	wal.buffer.Reset()
+	return wal.advanceFlushLSN()
+}
 
-	// Prevent memory bloat in long-running processes.
-	if wal.buffer.Cap() > wal.bufferSize*maximumBufferSizeFactor {
-		wal.buffer = wal.newBuffer()
+// advanceFlushLSN moves the flush position forward. In durable-ack mode it only
+// advances up to data the handler reports as durably persisted by the server,
+// which may lag the written position; otherwise it tracks the written position.
+// It is also called on idle flush ticks so that acknowledgements arriving after
+// the last write are still reflected in the flush position.
+func (wal *Data) advanceFlushLSN() error {
+	if !wal.requireDurableAck {
+		wal.flushLSN = wal.writeLSN
+		return nil
 	}
 
-	wal.flushLSN = wal.writeLSN
+	synced, err := wal.handler.SyncedOffset()
+	if err != nil {
+		return fmt.Errorf("while reading the durable offset: %w", err)
+	}
+
+	durableLSN := wal.fileStartLSN + synced
+	if durableLSN > wal.flushLSN {
+		wal.flushLSN = durableLSN
+	}
 
 	return nil
 }
@@ -224,5 +257,10 @@ func (wal *Data) closeCurrentWAL(ctx context.Context) error {
 		return fmt.Errorf("while closing current WAL file: %w", err)
 	}
 
+	// Closing the WAL file drains and verifies every outstanding
+	// acknowledgement, so the whole segment is now durable: advance the flush
+	// position to the segment boundary regardless of the mode.
+	wal.flushLSN = wal.writeLSN
+
 	return nil
 }
diff --git a/core/internal/client/sendwal/buffer/buffer_test.go b/core/internal/client/sendwal/buffer/buffer_test.go
new file mode 100644
index 00000000..f94152f9
--- /dev/null
+++ b/core/internal/client/sendwal/buffer/buffer_test.go
@@ -0,0 +1,179 @@
+/*
+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 (
+	"context"
+	"testing"
+
+	"github.com/cloudnative-pg/machinery/pkg/types"
+)
+
+// fakeHandler is a Handler whose durable offset can be controlled by the test,
+// letting it stand in for a Klio server that acknowledges durability lazily.
+type fakeHandler struct {
+	opened  bool
+	written uint64
+	synced  uint64
+}
+
+func (f *fakeHandler) HasWALFileOpened() bool { return f.opened }
+
+func (f *fakeHandler) OpenWAL(_ context.Context, _ uint64) error {
+	f.opened = true
+	f.written = 0
+	f.synced = 0
+
+	return nil
+}
+
+func (f *fakeHandler) CloseWAL(_ context.Context) error {
+	f.opened = false
+	return nil
+}
+
+func (f *fakeHandler) CurrentOffset() (uint64, error) { return f.written, nil }
+
+func (f *fakeHandler) Write(_ context.Context, p []byte) (int, error) {
+	f.written += uint64(len(p))
+	return len(p), nil
+}
+
+func (f *fakeHandler) SyncedOffset() (uint64, error) { return f.synced, nil }
+
+const testSegmentSize = 16 * 1024 * 1024
+
+// TestFlushLSNTracksWriteWhenDurableAckDisabled verifies that, in the default
+// mode, the flush position advances to the written position as soon as data is
+// handed to the send buffer.
+func TestFlushLSNTracksWriteWhenDurableAckDisabled(t *testing.T) {
+	ctx := context.Background()
+	handler := &fakeHandler{}
+	data := New(1, testSegmentSize, handler, 64*1024, false)
+
+	if err := data.ProcessWALData(ctx, make([]byte, 1000), types.LSN("0/0")); err != nil {
+		t.Fatalf("processing WAL data: %v", err)
+	}
+	if err := data.Flush(ctx); err != nil {
+		t.Fatalf("flushing: %v", err)
+	}
+
+	if data.WriteLSN() != 1000 {
+		t.Fatalf("expected write LSN 1000, got %d", data.WriteLSN())
+	}
+	if data.FlushLSN() != 1000 {
+		t.Fatalf("expected flush LSN to track write LSN (1000), got %d", data.FlushLSN())
+	}
+}
+
+// TestFlushLSNGatedOnDurableAck verifies that, when durable acknowledgements are
+// required, the flush position only advances up to the offset the handler
+// reports as durable, never ahead of it.
+func TestFlushLSNGatedOnDurableAck(t *testing.T) {
+	ctx := context.Background()
+	handler := &fakeHandler{}
+	data := New(1, testSegmentSize, handler, 64*1024, true)
+
+	if err := data.ProcessWALData(ctx, make([]byte, 1000), types.LSN("0/0")); err != nil {
+		t.Fatalf("processing WAL data: %v", err)
+	}
+
+	// Nothing acknowledged yet: the write position moves, the flush position
+	// must not.
+	if err := data.Flush(ctx); err != nil {
+		t.Fatalf("flushing: %v", err)
+	}
+	if data.WriteLSN() != 1000 {
+		t.Fatalf("expected write LSN 1000, got %d", data.WriteLSN())
+	}
+	if data.FlushLSN() != 0 {
+		t.Fatalf("expected flush LSN 0 before any ack, got %d", data.FlushLSN())
+	}
+
+	// A partial acknowledgement advances the flush position up to the durable
+	// offset only.
+	handler.synced = 600
+	if err := data.Flush(ctx); err != nil {
+		t.Fatalf("flushing: %v", err)
+	}
+	if data.FlushLSN() != 600 {
+		t.Fatalf("expected flush LSN 600 after partial ack, got %d", data.FlushLSN())
+	}
+
+	// Full acknowledgement lets the flush position reach the write position.
+	handler.synced = 1000
+	if err := data.Flush(ctx); err != nil {
+		t.Fatalf("flushing: %v", err)
+	}
+	if data.FlushLSN() != 1000 {
+		t.Fatalf("expected flush LSN 1000 after full ack, got %d", data.FlushLSN())
+	}
+}
+
+// TestFlushLSNNeverRegresses verifies that a smaller durable offset reported
+// afterwards cannot move the flush position backwards.
+func TestFlushLSNNeverRegresses(t *testing.T) {
+	ctx := context.Background()
+	handler := &fakeHandler{}
+	data := New(1, testSegmentSize, handler, 64*1024, true)
+
+	if err := data.ProcessWALData(ctx, make([]byte, 1000), types.LSN("0/0")); err != nil {
+		t.Fatalf("processing WAL data: %v", err)
+	}
+
+	handler.synced = 800
+	if err := data.Flush(ctx); err != nil {
+		t.Fatalf("flushing: %v", err)
+	}
+	if data.FlushLSN() != 800 {
+		t.Fatalf("expected flush LSN 800, got %d", data.FlushLSN())
+	}
+
+	handler.synced = 500
+	if err := data.Flush(ctx); err != nil {
+		t.Fatalf("flushing: %v", err)
+	}
+	if data.FlushLSN() != 800 {
+		t.Fatalf("expected flush LSN to stay at 800, got %d", data.FlushLSN())
+	}
+}
+
+// TestFlushLSNPinnedOnSegmentBoundary verifies that completing a WAL segment
+// advances the flush position to the segment boundary, because closing the file
+// drains and verifies every outstanding acknowledgement.
+func TestFlushLSNPinnedOnSegmentBoundary(t *testing.T) {
+	ctx := context.Background()
+	handler := &fakeHandler{}
+	const smallSegment = 2048
+	data := New(1, smallSegment, handler, 64*1024, true)
+
+	// Writing exactly one segment crosses the boundary and closes the file.
+	if err := data.ProcessWALData(ctx, make([]byte, smallSegment), types.LSN("0/0")); err != nil {
+		t.Fatalf("processing WAL data: %v", err)
+	}
+
+	if data.WriteLSN() != smallSegment {
+		t.Fatalf("expected write LSN %d, got %d", smallSegment, data.WriteLSN())
+	}
+	// Even though no ack was recorded, the closed segment is durable.
+	if data.FlushLSN() != smallSegment {
+		t.Fatalf("expected flush LSN pinned to boundary %d, got %d", smallSegment, data.FlushLSN())
+	}
+}
diff --git a/core/internal/client/sendwal/buffer/grpc.go b/core/internal/client/sendwal/buffer/grpc.go
index 6cf88658..95a78c29 100644
--- a/core/internal/client/sendwal/buffer/grpc.go
+++ b/core/internal/client/sendwal/buffer/grpc.go
@@ -118,3 +118,13 @@ func (wal *KlioClientStreamingHandler) Write(ctx context.Context, block []byte)
 
 	return len(block), nil
 }
+
+// SyncedOffset implements the Handler interface. It reports the number of bytes
+// of the current WAL file the Klio server has acknowledged as durable.
+func (wal *KlioClientStreamingHandler) SyncedOffset() (uint64, error) {
+	if wal.stream == nil {
+		return 0, nil
+	}
+
+	return wal.stream.SyncedOffset()
+}
diff --git a/core/internal/client/sendwal/buffer/handler.go b/core/internal/client/sendwal/buffer/handler.go
index d17859a4..f5555557 100644
--- a/core/internal/client/sendwal/buffer/handler.go
+++ b/core/internal/client/sendwal/buffer/handler.go
@@ -39,4 +39,9 @@ type Handler interface {
 
 	// Write writes data in the current WAL file
 	Write(ctx context.Context, p []byte) (n int, err error)
+
+	// SyncedOffset returns the number of bytes of the current WAL file that the
+	// destination has acknowledged as durably persisted. For handlers that do
+	// not receive durability acknowledgements, this equals the written offset.
+	SyncedOffset() (uint64, error)
 }
diff --git a/core/internal/client/sendwal/buffer/memory.go b/core/internal/client/sendwal/buffer/memory.go
index 3021af04..6394a8ea 100644
--- a/core/internal/client/sendwal/buffer/memory.go
+++ b/core/internal/client/sendwal/buffer/memory.go
@@ -99,3 +99,9 @@ func (wal *MemBufferHandler) CurrentOffset() (uint64, error) {
 func (wal *MemBufferHandler) Write(_ context.Context, p []byte) (int, error) {
 	return wal.buffer.Write(p) //nolint:wrapcheck
 }
+
+// SyncedOffset implements the Handler interface. The in-memory handler has no
+// remote durability round-trip, so the synced offset equals the written one.
+func (wal *MemBufferHandler) SyncedOffset() (uint64, error) {
+	return wal.CurrentOffset()
+}
diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go
index 245d5a68..3696dc14 100644
--- a/core/internal/client/sendwal/receiver.go
+++ b/core/internal/client/sendwal/receiver.go
@@ -445,6 +445,7 @@ func (s *Process) startReplication(
 			walSegmentSize,
 			klioHandler,
 			s.config.Source.BufferSize,
+			s.config.Source.RequireDurableAck,
 		)
 
 		copyDoneResult, err := s.manageWALStream(ctx, conn, walBuffer)
diff --git a/core/internal/grpc/klio_wal.pb.go b/core/internal/grpc/klio_wal.pb.go
index 0ef95fb2..ffdcf422 100644
--- a/core/internal/grpc/klio_wal.pb.go
+++ b/core/internal/grpc/klio_wal.pb.go
@@ -117,9 +117,15 @@ func (x *PutRequest) GetSendToTier2() bool {
 	return false
 }
 
+// This is streamed back by the WAL server on the Put stream. Each message
+// acknowledges that the bytes up to written_size have been durably persisted
+// (fsynced) on the server. The client uses it to advance the flush position it
+// reports to PostgreSQL only up to data that is genuinely durable.
 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"`
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Cumulative number of bytes of the current WAL file that have been durably
+	// persisted (fsynced) on the server.
+	WrittenSize   uint64 `protobuf:"varint,1,opt,name=written_size,json=writtenSize,proto3" json:"written_size,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -934,9 +940,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/server/walserver/upload.go b/core/internal/server/walserver/upload.go
index bb37dac5..b2d35ede 100644
--- a/core/internal/server/walserver/upload.go
+++ b/core/internal/server/walserver/upload.go
@@ -222,6 +222,28 @@ func (h *putHandler) processBlock(ctx context.Context, request *grpc.PutRequest)
 	h.writtenSize += uint64(len(request.GetWalBlock()))
 	h.recordLatestWrittenLSN(ctx)
 
+	// The block has been fsynced by writeBlock, so acknowledge the newly
+	// durable position back to the client. This lets a client that requires
+	// durable acknowledgements advance the flush position it reports to
+	// PostgreSQL at a per-block granularity instead of once per whole segment.
+	return h.sendAck()
+}
+
+// sendAck reports the cumulative number of durably persisted bytes to the
+// client on the Put stream.
+func (h *putHandler) sendAck() error {
+	if err := h.req.Send(&grpc.PutResult{WrittenSize: h.writtenSize}); err != nil {
+		h.logger.Warning(
+			"Error while sending WAL Put acknowledgement",
+			"writtenSize", h.writtenSize,
+			"walFileName", h.blockMeta.walFileName,
+			"clusterName", h.blockMeta.clusterName,
+			"err", err,
+		)
+
+		return status.Errorf(grpccodes.Internal, "error while sending acknowledgement: %v", err.Error())
+	}
+
 	return nil
 }
 
@@ -417,18 +439,11 @@ func (h *putHandler) finalize(ctx context.Context) error {
 		return err
 	}
 
-	if err := h.req.SendAndClose(&grpc.PutResult{
-		WrittenSize: h.writtenSize,
-	}); err != nil {
-		h.logger.Warning(
-			"Error while sending WAL Put response",
-			"writtenSize", h.writtenSize,
-			"walFileName", h.blockMeta.walFileName,
-			"clusterName", h.blockMeta.clusterName,
-			"err", err,
-		)
-
-		return status.Errorf(grpccodes.Internal, "error while sending response: %v", err.Error())
+	// Send a terminal acknowledgement carrying the final durable size. This
+	// mirrors the last per-block acknowledgement and guarantees the client
+	// observes the complete size before the stream is closed.
+	if err := h.sendAck(); err != nil {
+		return err
 	}
 
 	return h.notifyTier2(ctx)
@@ -436,12 +451,10 @@ 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,
-	}); err != nil {
+	if err := h.sendAck(); err != nil {
 		h.logger.Error(err, "Error while closing empty WAL file")
 
-		return status.Errorf(grpccodes.Internal, "error while closing (partial) WAL: %v", err.Error())
+		return err
 	}
 
 	return nil
diff --git a/core/pkg/config/client.go b/core/pkg/config/client.go
index a6726d59..f618e30a 100644
--- a/core/pkg/config/client.go
+++ b/core/pkg/config/client.go
@@ -90,6 +90,14 @@ type SourceConfig struct {
 	// 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"`
+
+	// RequireDurableAck makes the WAL receiver advance the flush position it
+	// reports to PostgreSQL only up to data the Klio server has acknowledged as
+	// durably persisted (fsynced). When false (the default), the flush position
+	// tracks data handed to the send buffer, which is faster but does not
+	// guarantee durability on the server. Enable this when Klio is used as a
+	// synchronous replication target.
+	RequireDurableAck bool `json:"require_durable_ack" mapstructure:"require_durable_ack"`
 }
 
 // ClientConfig is the configuration of the Klio client.
diff --git a/core/proto/klio_wal.proto b/core/proto/klio_wal.proto
index afbc3c15..9e083b2f 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) {}
@@ -47,7 +47,13 @@ message PutRequest {
   bool send_to_tier2 = 7;
 }
 
+// This is streamed back by the WAL server on the Put stream. Each message
+// acknowledges that the bytes up to written_size have been durably persisted
+// (fsynced) on the server. The client uses it to advance the flush position it
+// reports to PostgreSQL only up to data that is genuinely durable.
 message PutResult {
+  // Cumulative number of bytes of the current WAL file that have been durably
+  // persisted (fsynced) on the server.
   uint64 written_size = 1;
 }
 
diff --git a/documentation/web/docs/developer/_protocol.md b/documentation/web/docs/developer/_protocol.md
index 4a1665e2..aa5b1e1f 100644
--- a/documentation/web/docs/developer/_protocol.md
+++ b/documentation/web/docs/developer/_protocol.md
@@ -414,12 +414,15 @@ file
 <a name="klio-wal-v1-PutResult"></a>
 
 ### PutResult
-
+This is streamed back by the WAL server on the Put stream. Each message
+acknowledges that the bytes up to written_size have been durably persisted
+(fsynced) on the server. The client uses it to advance the flush position it
+reports to PostgreSQL only up to data that is genuinely durable.
 
 
 | Field | Type | Label | Description |
 | ----- | ---- | ----- | ----------- |
-| written_size | [uint64](#uint64) |  |  |
+| written_size | [uint64](#uint64) |  | Cumulative number of bytes of the current WAL file that have been durably persisted (fsynced) on the server. |
 
 
 
@@ -539,7 +542,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/pkg/config/client.go b/operator/pkg/config/client.go
index a6726d59..f618e30a 100644
--- a/operator/pkg/config/client.go
+++ b/operator/pkg/config/client.go
@@ -90,6 +90,14 @@ type SourceConfig struct {
 	// 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"`
+
+	// RequireDurableAck makes the WAL receiver advance the flush position it
+	// reports to PostgreSQL only up to data the Klio server has acknowledged as
+	// durably persisted (fsynced). When false (the default), the flush position
+	// tracks data handed to the send buffer, which is faster but does not
+	// guarantee durability on the server. Enable this when Klio is used as a
+	// synchronous replication target.
+	RequireDurableAck bool `json:"require_durable_ack" mapstructure:"require_durable_ack"`
 }
 
 // ClientConfig is the configuration of the Klio client.

From cf104672373c564462c30692f2dcc4a862b9f7a0 Mon Sep 17 00:00:00 2001
From: Gabriele Quaresima <gabriele.quaresima@enterprisedb.com>
Date: Mon, 17 Aug 2026 11:00:36 +0200
Subject: [PATCH 2/2] feat(operator): expose requireDurableAck in
 PluginConfiguration

Add a requireDurableAck field to the PluginConfiguration spec and wire it into
the generated WAL client configuration, so synchronous replication targets can
require durable acknowledgements. Document how to enable it.

Assisted-by: Claude
Signed-off-by: Gabriele Quaresima <gabriele.quaresima@enterprisedb.com>
---
 documentation/.wordlist.txt                   |  3 ++
 documentation/web/docs/user/api/_klio_api.md  |  1 +
 documentation/web/docs/user/wal_streaming.md  | 54 +++++++++++++++++--
 .../v1alpha1/plugin_configuration_types.go    |  9 ++++
 .../klio.cnpg.io_pluginconfigurations.yaml    |  9 ++++
 .../chart/crds/pluginconfiguration-crd.yaml   |  9 ++++
 operator/internal/klioconfig/config.go        |  7 +--
 7 files changed, 86 insertions(+), 6 deletions(-)

diff --git a/documentation/.wordlist.txt b/documentation/.wordlist.txt
index 8d9adc9f..2bd6ce94 100644
--- a/documentation/.wordlist.txt
+++ b/documentation/.wordlist.txt
@@ -145,6 +145,7 @@ RefreshResult
 RequestWALStart
 RequestWALStartRequest
 RequestWALStartResult
+RequireDurableAck
 ResetWALStream
 ResetWALStreamRequest
 ResetWALStreamResult
@@ -235,6 +236,8 @@ env
 failover
 fileReference
 filesystem
+fsync
+fsynced
 fullnameOverride
 gRPC
 ghcr
diff --git a/documentation/web/docs/user/api/_klio_api.md b/documentation/web/docs/user/api/_klio_api.md
index d666a314..082bc1cb 100644
--- a/documentation/web/docs/user/api/_klio_api.md
+++ b/documentation/web/docs/user/api/_klio_api.md
@@ -153,6 +153,7 @@ _Appears in:_
 | `tier1` _[Tier1PluginConfiguration](#tier1pluginconfiguration)_ | Tier1 is the Tier 1 configuration |  |  | Optional: \{\} <br /> |
 | `tier2` _[Tier2PluginConfiguration](#tier2pluginconfiguration)_ | Tier2 is the Tier 2 configuration |  |  | Optional: \{\} <br /> |
 | `walPrefetch` _[WALPrefetchConfiguration](#walprefetchconfiguration)_ | WALPrefetch configures WAL prefetching behavior during recovery operations. |  |  | Optional: \{\} <br /> |
+| `requireDurableAck` _boolean_ | RequireDurableAck makes the WAL streamer advance the flush position it<br />reports to PostgreSQL only up to WAL data the Klio server has confirmed as<br />durably persisted (fsynced). Enable it when using Klio as a synchronous<br />replication target for zero RPO. When false (the default), the flush<br />position tracks data handed to the send buffer, which is faster but does<br />not guarantee durability on the server. |  |  | Optional: \{\} <br /> |
 | `clientSecretName` _string_ | ClientSecretName is the name of the secret containing the client credentials | True |  | MinLength: 1 <br />Required: \{\} <br /> |
 | `serverSecretName` _string_ | ServerSecretName is the name of the secret containing the server TLS certificate | True |  | MinLength: 1 <br />Required: \{\} <br /> |
 | `clusterName` _string_ | ClusterName is the name of the PostgreSQL cluster we are connecting to | True |  | MinLength: 1 <br />Required: \{\} <br /> |
diff --git a/documentation/web/docs/user/wal_streaming.md b/documentation/web/docs/user/wal_streaming.md
index 4734b361..d797c35d 100644
--- a/documentation/web/docs/user/wal_streaming.md
+++ b/documentation/web/docs/user/wal_streaming.md
@@ -10,7 +10,9 @@ streaming for PostgreSQL. This architecture enables:
 - Partial WAL segment streaming, ensuring real-time data transfer
 - Built-in compression and encryption using user-provided keys
 - Controlled replication slot advancement, protecting against WAL loss
-- Optional synchronous replication, offering zero RPO when enabled
+- Optional synchronous replication, offering zero RPO when durable
+  acknowledgments are required (see
+  [Durable flush feedback](#durable-flush-feedback))
 
 ## Architecture
 
@@ -46,7 +48,8 @@ approach:
 
 - **Near-zero RPO:** WAL changes are streamed incrementally in near real-time,
   reducing the worst-case recovery point objective (RPO) from 5 minutes to
-  near-zero, or even zero in synchronous mode.
+  near-zero, or even zero in synchronous mode when durable acknowledgments are
+  required (see [Durable flush feedback](#durable-flush-feedback)).
 
 - **Improved efficiency and scalability:** A single, continuously running WAL
   streamer process replaces the need to spawn a new process for each WAL
@@ -113,8 +116,53 @@ explanation of the key fields:
   streaming).
 - `sent_lsn`, `write_lsn`, `flush_lsn`, `replay_lsn`: Positions in the WAL
   indicating how far data has been sent, written, flushed, and replayed on the
-  Klio server (replayed and flushed are always identical).
+  Klio server (replayed and flushed are always identical). By default,
+  `flush_lsn` tracks data handed to the WAL send buffer; when durable
+  acknowledgments are required, it only advances up to data the Klio server
+  has confirmed as durably persisted (see
+  [Durable flush feedback](#durable-flush-feedback)).
 - `write_lag`, `flush_lag`, `replay_lag`: Delays between WAL positions
   indicating replication latency.
 - `sync_state`: The synchronization state of this standby (e.g., `async`,
   `sync`, `potential`, `quorum`).
+
+## Durable flush feedback
+
+The flush position (`flush_lsn`) that the WAL streamer reports to PostgreSQL
+also drives two safety-critical decisions: it acknowledges commits when Klio is
+a synchronous standby, and it governs how far the physical replication slot's
+`restart_lsn` advances, which controls when PostgreSQL may recycle WAL segments.
+
+By default, the WAL streamer advances the flush position as soon as WAL data is
+handed to its send buffer, before the Klio server confirms that the data is
+durable. This is the fastest option and is appropriate for asynchronous WAL
+archiving, where a small window of not-yet-durable data is acceptable.
+
+When Klio is used as a **synchronous replication target** for zero RPO
+(`synchronous_standby_names` includes `klio` and `synchronous_commit` is set to
+`remote_flush` or `on`), this default is not safe: PostgreSQL could acknowledge
+a commit as durable, or recycle a WAL segment, before the corresponding bytes
+are guaranteed to survive a Klio server or network failure.
+
+For these deployments, set `requireDurableAck` to `true` in the
+`PluginConfiguration`. The Klio server then acknowledges each WAL block once it
+has been fsynced, and the WAL streamer only advances the flush position up to
+data that has been durably persisted:
+
+```yaml
+apiVersion: klio.cnpg.io/v1alpha1
+kind: PluginConfiguration
+metadata:
+  name: my-config
+spec:
+  serverAddress: klio-server.default
+  clientSecretName: my-client-credentials
+  serverSecretName: klio-server-tls
+  clusterName: my-cluster
+  requireDurableAck: true
+```
+
+Requiring durable acknowledgments adds a per-flush latency equal to the
+round-trip time to the Klio server plus its fsync time, so the flush position
+lags the write position by that amount. WAL upload throughput is unaffected,
+because WAL blocks are still sent without waiting for each acknowledgment.
diff --git a/operator/api/v1alpha1/plugin_configuration_types.go b/operator/api/v1alpha1/plugin_configuration_types.go
index 83d07b7d..cdd9f153 100644
--- a/operator/api/v1alpha1/plugin_configuration_types.go
+++ b/operator/api/v1alpha1/plugin_configuration_types.go
@@ -57,6 +57,15 @@ type PluginConfigurationSpec struct {
 	// +optional
 	WALPrefetch *WALPrefetchConfiguration `json:"walPrefetch,omitempty"`
 
+	// RequireDurableAck makes the WAL streamer advance the flush position it
+	// reports to PostgreSQL only up to WAL data the Klio server has confirmed as
+	// durably persisted (fsynced). Enable it when using Klio as a synchronous
+	// replication target for zero RPO. When false (the default), the flush
+	// position tracks data handed to the send buffer, which is faster but does
+	// not guarantee durability on the server.
+	// +optional
+	RequireDurableAck bool `json:"requireDurableAck,omitempty"`
+
 	// ClientSecretName is the name of the secret containing the client credentials
 	// +kubebuilder:validation:Required
 	// +kubebuilder:validation:MinLength=1
diff --git a/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml b/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml
index 830fbbc1..4158d723 100644
--- a/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml
+++ b/operator/config/crd/bases/klio.cnpg.io_pluginconfigurations.yaml
@@ -1603,6 +1603,15 @@ spec:
               pprof:
                 description: Pprof enables the pprof endpoint for performance profiling
                 type: boolean
+              requireDurableAck:
+                description: |-
+                  RequireDurableAck makes the WAL streamer advance the flush position it
+                  reports to PostgreSQL only up to WAL data the Klio server has confirmed as
+                  durably persisted (fsynced). Enable it when using Klio as a synchronous
+                  replication target for zero RPO. When false (the default), the flush
+                  position tracks data handed to the send buffer, which is faster but does
+                  not guarantee durability on the server.
+                type: boolean
               serverAddress:
                 description: ServerAddress is the address of the Klio server
                 minLength: 1
diff --git a/operator/dist/chart/crds/pluginconfiguration-crd.yaml b/operator/dist/chart/crds/pluginconfiguration-crd.yaml
index 3b2e1408..560807d6 100644
--- a/operator/dist/chart/crds/pluginconfiguration-crd.yaml
+++ b/operator/dist/chart/crds/pluginconfiguration-crd.yaml
@@ -1602,6 +1602,15 @@ spec:
               pprof:
                 description: Pprof enables the pprof endpoint for performance profiling
                 type: boolean
+              requireDurableAck:
+                description: |-
+                  RequireDurableAck makes the WAL streamer advance the flush position it
+                  reports to PostgreSQL only up to WAL data the Klio server has confirmed as
+                  durably persisted (fsynced). Enable it when using Klio as a synchronous
+                  replication target for zero RPO. When false (the default), the flush
+                  position tracks data handed to the send buffer, which is faster but does
+                  not guarantee durability on the server.
+                type: boolean
               serverAddress:
                 description: ServerAddress is the address of the Klio server
                 minLength: 1
diff --git a/operator/internal/klioconfig/config.go b/operator/internal/klioconfig/config.go
index 89b43d0d..e874861e 100644
--- a/operator/internal/klioconfig/config.go
+++ b/operator/internal/klioconfig/config.go
@@ -100,9 +100,10 @@ func GenerateConfig(
 
 	klioConfig := &config.Data{
 		Source: config.SourceConfig{
-			DSN:         "user=postgres replication=yes application_name=klio",
-			StandardDSN: "user=postgres application_name=klio",
-			Slot:        "klio",
+			DSN:               "user=postgres replication=yes application_name=klio",
+			StandardDSN:       "user=postgres application_name=klio",
+			Slot:              "klio",
+			RequireDurableAck: spec.RequireDurableAck,
 			// The following parameters are not used by the plugin, but here with their default for completeness
 			StandbyMessageTimeoutSeconds: 0,
 			FlushTimeoutMilliseconds:     0,