Test wal sender feedback

core/cmd/backup/delete.go

@@ -47,9 +47,6 @@ var deleteCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the default values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

core/cmd/backup/get-metadata.go

@@ -53,9 +53,6 @@ func runGetMetadata(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("could not unmarshal configuration: %w", err)
 	}
 
-	// Sets the default values, to be overridden by the user configuration
-	configuration.SetDefaults()
-
 	if configuration.Client == (config.ClientConfig{}) {
 		return cli.ErrClientSectionIsRequired
 	}

core/cmd/backup/list.go

@@ -47,9 +47,6 @@ var listCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the default values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

core/cmd/backup/run.go

@@ -61,9 +61,6 @@ func runBackup(cmd *cobra.Command, _ []string) error {
 		return fmt.Errorf("could not unmarshal configuration: %w", err)
 	}
 
-	// Sets the default values, to be overridden by the user configuration
-	configuration.SetDefaults()
-
 	if configuration.Client == (config.ClientConfig{}) {
 		return cli.ErrClientSectionIsRequired
 	}

core/cmd/backup/verify.go

@@ -146,9 +146,6 @@ func loadAndValidateConfig() (*config.Data, error) {
 		return nil, fmt.Errorf("could not unmarshal configuration: %w", err)
 	}
 
-	// Sets the default values, to be overridden by the user configuration
-	configuration.SetDefaults()
-
 	if configuration.Client == (config.ClientConfig{}) {
 		return nil, cli.ErrClientSectionIsRequired
 	}

core/cmd/get-metadata.go

@@ -49,9 +49,6 @@ var getMetadataCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the defaults values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

core/cmd/get-wal.go

@@ -56,9 +56,6 @@ var getWalCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the defaults values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

core/cmd/reset-lsn.go

@@ -49,9 +49,6 @@ var resetLSNCommand = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the defaults values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Source == (config.SourceConfig{}) {
 			return cli.ErrSourceSectionIsRequired
 		}

core/cmd/restore.go

@@ -62,9 +62,6 @@ var restoreCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the default values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

core/cmd/retention/get.go

@@ -47,9 +47,6 @@ var getCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the default values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

core/cmd/retention/set.go

@@ -47,9 +47,6 @@ var setCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the defaults values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

core/cmd/send-wal.go

@@ -67,9 +67,6 @@ var sendWalCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the defaults values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Source == (config.SourceConfig{}) {
 			return cli.ErrSourceSectionIsRequired
 		}

core/cmd/walplayer/play.go

@@ -54,9 +54,6 @@ var playCmd = &cobra.Command{
 			return fmt.Errorf("could not unmarshal configuration: %w", err)
 		}
 
-		// Sets the defaults values, to be overridden by the user configuration
-		configuration.SetDefaults()
-
 		if configuration.Client == (config.ClientConfig{}) {
 			return cli.ErrClientSectionIsRequired
 		}

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

@@ -46,18 +46,7 @@ type grpcWALStream struct {
 
 // Close implements common.WALStream.
 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 result.GetWrittenSize() != g.sentBytes {
-		return &IncompleteWALFileError{
-			uploadedSize: result.GetWrittenSize(),
-			expectedSize: g.sentBytes,
-		}
-	}
-
+	g.innerStream.CloseSend()
 	return nil
 }
 

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

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

@@ -38,8 +38,8 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte,
 	}
 
 	walReader := bytes.NewBuffer(content)
-
 	buffer := make([]byte, 4096)
+
 	for {
 		readBytes, readError := walReader.Read(buffer)
 		if readError != nil && !errors.Is(readError, io.EOF) {
@@ -56,23 +56,23 @@ func (c *Connection) StoreWAL(ctx context.Context, name string, content []byte,
 			return fmt.Errorf("error while sending WAL 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) {
 			break
 		}
 	}
 
-	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
 }
 

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

@@ -40,19 +40,19 @@ func (c *Connection) StoreWALStreaming(
 	name string,
 	segmentSize uint64,
 	sendToTier2 bool,
-) (*klioclient.WALUploader, error) {
+) (klioclient.WALUploaderImpl, 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{
+	return &grpcWALStream{
 		innerStream: stream,
 		segmentSize: segmentSize,
 		clusterName: c.clientConfig.ClusterName,
 		walName:     name,
 		sendToTier2: sendToTier2,
-	}), nil
+	}, nil
 }
 
 // GetWALStreaming get a WAL from a remote connection.

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

@@ -51,3 +51,14 @@ func (g *grpcWALStream) SendBlock(ctx context.Context, block []byte) error {
 
 	return nil
 }
+
+// ReceiveFeedback receives the feedback from the server, that is telling us how
+// many bytes it flushed.
+func (g *grpcWALStream) ReceiveFeedback(ctx context.Context) (uint64, error) {
+	result, err := g.innerStream.Recv()
+	if err != nil {
+		return 0, err
+	}
+
+	return result.GetWrittenSize(), nil
+}

core/internal/client/klioclient/interfaces.go

@@ -103,28 +103,10 @@ type WALUploaderImpl interface {
 	// SendBlock sends a WAL Block
 	SendBlock(ctx context.Context, block []byte) error
 
+	// ReceiveFeedback receives a feedback from the server that is telling
+	// us how many bytes it has flushed.
+	ReceiveFeedback(ctx context.Context) (uint64, 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
-}