feat(core): extract WAL sender into a Klio-decoupled pkg/sendwal package

core/cmd/reset-lsn.go

@@ -28,8 +28,8 @@ import (
 
 	"github.com/cloudnative-pg/klio/core/internal/cli"
 	"github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient"
-	"github.com/cloudnative-pg/klio/core/internal/client/sendwal"
 	"github.com/cloudnative-pg/klio/core/pkg/config"
+	"github.com/cloudnative-pg/klio/core/pkg/sendwal"
 )
 
 // resetLSNCommand represents the run command
@@ -73,9 +73,22 @@ var resetLSNCommand = &cobra.Command{
 			return fmt.Errorf("while connecting to the Klio server: %w", err)
 		}
 
-		return sendwal.
-			New(&configuration, contextLogger, client, false).
-			ResetReplicationStatus(cmd.Context())
+		coordinator := grpcclient.NewSendWALCoordinator(client, false)
+		handlerFactory := grpcclient.NewKlioClientHandlerFactory(client, false)
+
+		return sendwal.New(
+			configuration.Source.DSN,
+			contextLogger,
+			coordinator,
+			handlerFactory,
+			sendwal.Options{
+				Slot:                  configuration.Source.Slot,
+				ClusterName:           configuration.Client.ClusterName,
+				BufferSize:            configuration.Source.BufferSize,
+				FlushTimeout:          configuration.Source.FlushTimeout(),
+				StandbyMessageTimeout: configuration.Source.StandbyMessageTimeout(),
+			},
+		).ResetReplicationStatus(cmd.Context())
 	},
 }
 

core/cmd/send-wal.go

@@ -35,9 +35,9 @@ import (
 
 	"github.com/cloudnative-pg/klio/core/internal/cli"
 	"github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient"
-	"github.com/cloudnative-pg/klio/core/internal/client/sendwal"
 	"github.com/cloudnative-pg/klio/core/internal/opentelemetry"
 	"github.com/cloudnative-pg/klio/core/pkg/config"
+	"github.com/cloudnative-pg/klio/core/pkg/sendwal"
 )
 
 // ErrTimeoutWaitingPG is raised when we couldn't get a connection to PostgreSQL.
@@ -96,8 +96,22 @@ var sendWalCmd = &cobra.Command{
 			return fmt.Errorf("while connecting to the Klio server: %w", err)
 		}
 
-		err = sendwal.New(&configuration, logger, client, configuration.Tier2BackupEnabled).
-			Start(ctx)
+		coordinator := grpcclient.NewSendWALCoordinator(client, configuration.Tier2BackupEnabled)
+		handlerFactory := grpcclient.NewKlioClientHandlerFactory(client, configuration.Tier2BackupEnabled)
+
+		err = sendwal.New(
+			configuration.Source.DSN,
+			logger,
+			coordinator,
+			handlerFactory,
+			sendwal.Options{
+				Slot:                  configuration.Source.Slot,
+				ClusterName:           configuration.Client.ClusterName,
+				BufferSize:            configuration.Source.BufferSize,
+				FlushTimeout:          configuration.Source.FlushTimeout(),
+				StandbyMessageTimeout: configuration.Source.StandbyMessageTimeout(),
+			},
+		).Start(ctx)
 		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 			logger.Info("send-wal stopped due to context cancellation, exiting gracefully")
 			return nil

core/internal/client/klioclient/grpcclient/sendwal_adapter.go

@@ -0,0 +1,83 @@
+/*
+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 grpcclient
+
+import (
+	"context"
+	"fmt"
+
+	klioGRPC "github.com/cloudnative-pg/klio/core/internal/grpc"
+)
+
+// SendWALCoordinator adapts a *Connection to the sendwal.ReplicationCoordinator
+// interface (github.com/cloudnative-pg/klio/core/pkg/sendwal), so the
+// generic WAL receiver can negotiate streaming with a Klio server without
+// that package knowing anything about Klio's gRPC protocol.
+type SendWALCoordinator struct {
+	conn        *Connection
+	sendToTier2 bool
+}
+
+// NewSendWALCoordinator creates a new SendWALCoordinator.
+func NewSendWALCoordinator(conn *Connection, sendToTier2 bool) *SendWALCoordinator {
+	return &SendWALCoordinator{
+		conn:        conn,
+		sendToTier2: sendToTier2,
+	}
+}
+
+// RequestStart implements sendwal.ReplicationCoordinator.
+func (c *SendWALCoordinator) RequestStart(
+	ctx context.Context,
+	clusterName, systemID, currentWALName string,
+) (string, error) {
+	result, err := c.conn.RequestWALStart(ctx, &klioGRPC.RequestWALStartRequest{
+		ClusterName:    clusterName,
+		SystemId:       systemID,
+		CurrentWalName: currentWALName,
+	})
+	if err != nil {
+		return "", fmt.Errorf("while requesting WAL start: %w", err)
+	}
+
+	return result.GetWalName(), nil
+}
+
+// ResetStream implements sendwal.ReplicationCoordinator.
+func (c *SendWALCoordinator) ResetStream(
+	ctx context.Context,
+	clusterName, systemID, currentWALName string,
+) (string, error) {
+	result, err := c.conn.ResetWALStream(ctx, &klioGRPC.ResetWALStreamRequest{
+		ClusterName:    clusterName,
+		SystemId:       systemID,
+		CurrentWalName: currentWALName,
+	})
+	if err != nil {
+		return "", fmt.Errorf("while resetting WAL stream: %w", err)
+	}
+
+	return result.GetWalName(), nil
+}
+
+// StoreHistoryFile implements sendwal.ReplicationCoordinator.
+func (c *SendWALCoordinator) StoreHistoryFile(ctx context.Context, name string, content []byte) error {
+	return c.conn.StoreHistoryFile(ctx, name, content, c.sendToTier2) //nolint:wrapcheck
+}

core/internal/client/klioclient/grpcclient/sendwal_adapter_test.go

@@ -0,0 +1,145 @@
+/*
+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 grpcclient
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudnative-pg/klio/core/internal/repository"
+	"github.com/cloudnative-pg/klio/core/pkg/config"
+)
+
+// newTestConnection creates a connection to a temporary, local Klio server
+// backed by an in-memory Kopia repository, closing it automatically once the
+// test finishes. This mirrors the setup already used by
+// BenchmarkLookupSnapshotsViaKlioServer in connection_test.go.
+func newTestConnection(ctx context.Context, t *testing.T, clusterName string) *Connection {
+	t.Helper()
+
+	conn, err := ConnectTemporary(
+		ctx,
+		log.GetLogger(),
+		&config.ClientConfig{
+			ClusterName: clusterName,
+		},
+		repository.Options{
+			FS:       afero.NewMemMapFs(),
+			Password: "random-string",
+		},
+	)
+	require.NoError(t, err, "while creating temporary Klio repository")
+
+	t.Cleanup(func() {
+		require.NoError(t, conn.Close())
+	})
+
+	return &conn.Connection
+}
+
+func TestSendWALCoordinatorRequestStart(t *testing.T) {
+	ctx := t.Context()
+
+	t.Run("first contact, no prior metadata for the cluster", func(t *testing.T) {
+		conn := newTestConnection(ctx, t, "cluster-name")
+		coordinator := NewSendWALCoordinator(conn, false)
+
+		// GIVEN a cluster the destination has never seen before
+		// WHEN the source requests a start position
+		walName, err := coordinator.RequestStart(ctx, "first-contact-cluster", "system-1", "000000010000000000000001")
+
+		// THEN the destination has nothing to negotiate with, so it accepts
+		// the source's own current WAL file name as the start position
+		require.NoError(t, err)
+		assert.Equal(t, "000000010000000000000001", walName)
+	})
+
+	t.Run("system ID mismatch on a second request for the same cluster", func(t *testing.T) {
+		conn := newTestConnection(ctx, t, "cluster-name")
+		coordinator := NewSendWALCoordinator(conn, false)
+
+		// GIVEN a cluster that has already negotiated a start with system ID "system-a"
+		_, err := coordinator.RequestStart(ctx, "mismatch-cluster", "system-a", "000000010000000000000001")
+		require.NoError(t, err)
+
+		// WHEN a different system ID negotiates a start for the same cluster name
+		// THEN the destination rejects it, since it would mix WALs from two
+		// different PostgreSQL instances under the same cluster name
+		_, err = coordinator.RequestStart(ctx, "mismatch-cluster", "system-b", "000000010000000000000002")
+		require.Error(t, err)
+	})
+}
+
+func TestSendWALCoordinatorResetStream(t *testing.T) {
+	ctx := t.Context()
+	conn := newTestConnection(ctx, t, "reset-cluster")
+	coordinator := NewSendWALCoordinator(conn, false)
+
+	// GIVEN a cluster with an established start position and one archived WAL
+	_, err := coordinator.RequestStart(ctx, "reset-cluster", "system-1", "000000010000000000000005")
+	require.NoError(t, err)
+
+	require.NoError(t, coordinator.StoreHistoryFile(ctx, "000000010000000000000005", []byte("wal-content")))
+
+	t.Run("resetting past the latest archived WAL succeeds", func(t *testing.T) {
+		// WHEN the source asks to reset the stream to a WAL file more recent
+		// than what is archived
+		walName, err := coordinator.ResetStream(ctx, "reset-cluster", "system-1", "000000010000000000000009")
+
+		// THEN the destination accepts it and echoes back the requested WAL name
+		require.NoError(t, err)
+		assert.Equal(t, "000000010000000000000009", walName)
+	})
+
+	t.Run("resetting to or before the latest archived WAL fails", func(t *testing.T) {
+		// WHEN the source asks to reset the stream to a WAL file that is not
+		// more recent than what is already archived
+		_, err := coordinator.ResetStream(ctx, "reset-cluster", "system-1", "000000010000000000000005")
+
+		// THEN the destination refuses, since it would create a gap in the
+		// archived WAL sequence
+		require.Error(t, err)
+	})
+}
+
+func TestSendWALCoordinatorStoreHistoryFile(t *testing.T) {
+	ctx := t.Context()
+	conn := newTestConnection(ctx, t, "history-cluster")
+
+	t.Run("stores the content under the given name", func(t *testing.T) {
+		// The temporary test server has no tier2 queue wired up, so this
+		// exercises the tier1-only path; tier2 propagation is checked
+		// structurally below, without a live send.
+		coordinator := NewSendWALCoordinator(conn, false)
+
+		err := coordinator.StoreHistoryFile(ctx, "00000001.history", []byte("history-file-content"))
+		require.NoError(t, err)
+	})
+
+	t.Run("remembers the tier2 flag it was built with", func(t *testing.T) {
+		coordinator := NewSendWALCoordinator(conn, true)
+		assert.True(t, coordinator.sendToTier2)
+	})
+}

core/internal/client/sendwal/buffer/grpc.go → core/internal/client/klioclient/grpcclient/wal_handler.go

@@ -17,7 +17,7 @@ limitations under the License.
 SPDX-License-Identifier: Apache-2.0
 */
 
-package buffer
+package grpcclient
 
 import (
 	"context"
@@ -27,13 +27,14 @@ import (
 	"github.com/cloudnative-pg/machinery/pkg/types"
 
 	"github.com/cloudnative-pg/klio/core/internal/client/klioclient"
-	"github.com/cloudnative-pg/klio/core/internal/client/klioclient/grpcclient"
+	sendwalbuffer "github.com/cloudnative-pg/klio/core/pkg/sendwal/buffer"
 )
 
-// KlioClientStreamingHandler is a handler that streams directly to a
-// Klio server.
+// KlioClientStreamingHandler is a sendwalbuffer.Handler that streams WAL
+// data directly to a Klio server as it is received, rather than buffering
+// a whole WAL file before sending it.
 type KlioClientStreamingHandler struct {
-	conn *grpcclient.Connection
+	conn *Connection
 
 	stream klioclient.WALUploaderImpl
 	offset uint64
@@ -49,7 +50,7 @@ type KlioClientStreamingHandler struct {
 func NewKlioClientHandler(
 	tli int,
 	segmentSize uint64,
-	conn *grpcclient.Connection,
+	conn *Connection,
 	sendToTier2 bool,
 ) *KlioClientStreamingHandler {
 	return &KlioClientStreamingHandler{
@@ -61,7 +62,19 @@ func NewKlioClientHandler(
 	}
 }
 
-// OpenWAL implements the Handler interface.
+// NewKlioClientHandlerFactory returns a sendwal.HandlerFactory
+// (github.com/cloudnative-pg/klio/core/pkg/sendwal) building
+// KlioClientStreamingHandler instances that stream to conn.
+func NewKlioClientHandlerFactory(
+	conn *Connection,
+	sendToTier2 bool,
+) func(tli int, segmentSize uint64) sendwalbuffer.Handler {
+	return func(tli int, segmentSize uint64) sendwalbuffer.Handler {
+		return NewKlioClientHandler(tli, segmentSize, conn, sendToTier2)
+	}
+}
+
+// OpenWAL implements the buffer.Handler interface.
 func (wal *KlioClientStreamingHandler) OpenWAL(ctx context.Context, blockpos uint64) error {
 	currentWALFile, err := types.Int64ToLSN(blockpos).WALFileName(wal.tli, wal.segmentSize)
 	if err != nil {
@@ -81,12 +94,12 @@ func (wal *KlioClientStreamingHandler) OpenWAL(ctx context.Context, blockpos uin
 	return nil
 }
 
-// HasWALFileOpened implements the Handler interface.
+// HasWALFileOpened implements the buffer.Handler interface.
 func (wal *KlioClientStreamingHandler) HasWALFileOpened() bool {
 	return wal.currentWALFile != ""
 }
 
-// CloseWAL implements the Handler interface.
+// CloseWAL implements the buffer.Handler interface.
 func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error {
 	contextLogger := log.FromContext(ctx)
 
@@ -102,12 +115,12 @@ func (wal *KlioClientStreamingHandler) CloseWAL(ctx context.Context) error {
 	return nil
 }
 
-// CurrentOffset implements the Handler interface.
+// CurrentOffset implements the buffer.Handler interface.
 func (wal *KlioClientStreamingHandler) CurrentOffset() (uint64, error) {
 	return wal.offset, nil
 }
 
-// Write implements the Handler interface.
+// Write implements the buffer.Handler interface.
 func (wal *KlioClientStreamingHandler) Write(ctx context.Context, block []byte) (int, error) {
 	err := wal.stream.SendBlock(ctx, block)
 	if err != nil {