Repository: xataio/pgstream
Pull request #148 (2025-02-17 16:58:26)

Treat snapshot errors as strings for recording
The snapshot errors are used to be able to record the snapshot status in the snapshot requests store, to keep track of issues during snapshotting. This PR updates the snapshot level errors as strings instead of errors, so that they are properly stored, similarly to what's done for the table errors.

Tests for errors are added.


Patch:
diff --git a/pkg/snapshot/errors.go b/pkg/snapshot/errors.go
index 3d7db79..af00add 100644
--- a/pkg/snapshot/errors.go
+++ b/pkg/snapshot/errors.go
@@ -2,11 +2,14 @@
 
 package snapshot
 
-import "errors"
+import (
+	"errors"
+	"strings"
+)
 
 type Errors struct {
-	Snapshot error        `json:"snapshot,omitempty"`
-	Tables   []TableError `json:"tables,omitempty"`
+	SnapshotErrMsgs []string     `json:"snapshot,omitempty"`
+	Tables          []TableError `json:"tables,omitempty"`
 }
 
 type TableError struct {
@@ -14,22 +17,6 @@ type TableError struct {
 	ErrorMsg string `json:"error"`
 }
 
-func (e *Errors) Error() string {
-	var err error
-	if e.Snapshot != nil {
-		err = e.Snapshot
-	}
-
-	for _, table := range e.Tables {
-		if err == nil {
-			err = table
-			continue
-		}
-		err = errors.Join(err, table)
-	}
-	return err.Error()
-}
-
 func NewErrors(err error) *Errors {
 	if err == nil {
 		return nil
@@ -39,22 +26,36 @@ func NewErrors(err error) *Errors {
 		return snapshotErrs
 	}
 	return &Errors{
-		Snapshot: err,
+		SnapshotErrMsgs: []string{err.Error()},
 	}
 }
 
+func (e *Errors) Error() string {
+	if e == nil {
+		return ""
+	}
+	errMsgs := make([]string, 0, len(e.SnapshotErrMsgs))
+	errMsgs = append(errMsgs, e.SnapshotErrMsgs...)
+
+	for _, table := range e.Tables {
+		errMsgs = append(errMsgs, table.Error())
+	}
+	return strings.Join(errMsgs, ";")
+}
+
 func (e *Errors) AddSnapshotError(err error) {
 	if e == nil {
-		e = &Errors{}
-		e.AddSnapshotError(err)
 		return
 	}
-
-	if e.Snapshot == nil {
-		e.Snapshot = err
+	if e.SnapshotErrMsgs == nil {
+		e.SnapshotErrMsgs = []string{err.Error()}
 		return
 	}
-	e.Snapshot = errors.Join(e.Snapshot, err)
+	e.SnapshotErrMsgs = append(e.SnapshotErrMsgs, err.Error())
+}
+
+func (e *Errors) IsSnapshotError() bool {
+	return e != nil && len(e.SnapshotErrMsgs) > 0
 }
 
 func (e *Errors) IsTableError(table string) bool {
@@ -62,7 +63,7 @@ func (e *Errors) IsTableError(table string) bool {
 		return false
 	}
 
-	if e.Snapshot != nil {
+	if e.SnapshotErrMsgs != nil {
 		return true
 	}
 
@@ -99,5 +100,5 @@ func NewTableError(table string, err error) TableError {
 }
 
 func (e TableError) Error() string {
-	return "snapshot error for table " + e.Table + ": " + e.ErrorMsg
+	return e.Table + ": " + e.ErrorMsg
 }
diff --git a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
index a5769df..36978fd 100644
--- a/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
+++ b/pkg/snapshot/generator/postgres/data/pg_snapshot_generator.go
@@ -82,7 +82,7 @@ func (sg *SnapshotGenerator) CreateSnapshot(ctx context.Context, ss *snapshot.Sn
 	return sg.conn.ExecInTxWithOptions(ctx, func(tx pglib.Tx) error {
 		snapshotID, err := sg.exportSnapshot(ctx, tx)
 		if err != nil {
-			return &snapshot.Errors{Snapshot: err}
+			return &snapshot.Errors{SnapshotErrMsgs: []string{err.Error()}}
 		}
 
 		tableChan := make(chan string, len(ss.TableNames))
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder.go b/pkg/snapshot/generator/snapshot_generator_recorder.go
index 43dbb2a..5655363 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder.go
@@ -35,6 +35,7 @@ func (s *SnapshotRecorder) CreateSnapshot(ctx context.Context, ss *snapshot.Snap
 	if err != nil {
 		return snapshot.NewErrors(err)
 	}
+
 	// no tables to snapshot
 	if len(filteredTables) == 0 {
 		return nil
@@ -107,10 +108,9 @@ func (s *SnapshotRecorder) filterOutExistingSnapshots(ctx context.Context, schem
 	filteredTables := make([]string, 0, len(tables))
 	for _, table := range tables {
 		wildCardReq, wildcardFound := existingRequests[wildcard]
-
 		switch table {
 		case wildcard:
-			if wildcardFound {
+			if wildcardFound && !wildCardReq.Errors.IsSnapshotError() {
 				filteredTables = append(filteredTables, wildCardReq.Errors.GetFailedTables()...)
 				break
 			}


Test patch:
diff --git a/pkg/snapshot/errors_test.go b/pkg/snapshot/errors_test.go
new file mode 100644
index 0000000..bfee697
--- /dev/null
+++ b/pkg/snapshot/errors_test.go
@@ -0,0 +1,353 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package snapshot
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	err1 = "error1"
+	err2 = "error2"
+
+	table1 = "table1"
+	table2 = "table2"
+)
+
+var errTest = errors.New("oh noes")
+
+func Test_NewErrors(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  error
+
+		wantErr *Errors
+	}{
+		{
+			name: "nil error",
+			err:  nil,
+
+			wantErr: nil,
+		},
+		{
+			name: "normal error",
+			err:  errTest,
+
+			wantErr: &Errors{
+				SnapshotErrMsgs: []string{errTest.Error()},
+			},
+		},
+		{
+			name: "snapshot error",
+			err: &Errors{
+				SnapshotErrMsgs: []string{"test"},
+			},
+
+			wantErr: &Errors{
+				SnapshotErrMsgs: []string{"test"},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			gotErr := NewErrors(tc.err)
+			require.Equal(t, gotErr, tc.wantErr)
+		})
+	}
+}
+
+func TestErrors_Error(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  error
+
+		wantStr string
+	}{
+		{
+			name: "nil error",
+			err:  (*Errors)(nil),
+
+			wantStr: "",
+		},
+		{
+			name: "empty error",
+			err:  &Errors{},
+
+			wantStr: "",
+		},
+		{
+			name: "with one snapshot error",
+			err: &Errors{
+				SnapshotErrMsgs: []string{err1},
+			},
+
+			wantStr: err1,
+		},
+		{
+			name: "with multiple snapshot errors",
+			err: &Errors{
+				SnapshotErrMsgs: []string{err1, err2},
+			},
+
+			wantStr: fmt.Sprintf("%s;%s", err1, err2),
+		},
+		{
+			name: "with one table error",
+			err: &Errors{
+				Tables: []TableError{
+					{ErrorMsg: err1, Table: table1},
+				},
+			},
+
+			wantStr: fmt.Sprintf("%s: %s", table1, err1),
+		},
+		{
+			name: "with multiple table errors",
+			err: &Errors{
+				Tables: []TableError{
+					{ErrorMsg: err1, Table: table1},
+					{ErrorMsg: err2, Table: table2},
+				},
+			},
+
+			wantStr: fmt.Sprintf("%s: %s;%s: %s", table1, err1, table2, err2),
+		},
+		{
+			name: "with snapshot and table errors",
+			err: &Errors{
+				SnapshotErrMsgs: []string{err1, err2},
+				Tables: []TableError{
+					NewTableError(table1, errors.New(err1)),
+					NewTableError(table2, errors.New(err2)),
+				},
+			},
+
+			wantStr: fmt.Sprintf("%s;%s;%s: %s;%s: %s", err1, err2, table1, err1, table2, err2),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			errMsg := tc.err.Error()
+			require.Equal(t, tc.wantStr, errMsg)
+		})
+	}
+}
+
+func TestErrors_AddSnapshotErrors(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name   string
+		err    *Errors
+		addErr error
+
+		wantErr error
+	}{
+		{
+			name:   "nil error",
+			err:    nil,
+			addErr: errTest,
+
+			wantErr: (*Errors)(nil),
+		},
+		{
+			name:   "error",
+			err:    &Errors{},
+			addErr: errTest,
+
+			wantErr: &Errors{
+				SnapshotErrMsgs: []string{errTest.Error()},
+			},
+		},
+		{
+			name: "with snapshot errors",
+			err: &Errors{
+				SnapshotErrMsgs: []string{err1, err2},
+			},
+			addErr: errTest,
+
+			wantErr: &Errors{
+				SnapshotErrMsgs: []string{err1, err2, errTest.Error()},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tc.err.AddSnapshotError(tc.addErr)
+			require.Equal(t, tc.wantErr, tc.err)
+		})
+	}
+}
+
+func TestErrors_IsTableError(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name  string
+		table string
+		err   *Errors
+
+		wantTableErr bool
+	}{
+		{
+			name:  "nil error",
+			err:   nil,
+			table: table1,
+
+			wantTableErr: false,
+		},
+		{
+			name:  "no errors",
+			err:   &Errors{},
+			table: table1,
+
+			wantTableErr: false,
+		},
+		{
+			name: "with snapshto errors",
+			err: &Errors{
+				SnapshotErrMsgs: []string{err1},
+			},
+			table: table1,
+
+			wantTableErr: true,
+		},
+		{
+			name: "with table errors",
+			err: &Errors{
+				Tables: []TableError{
+					{Table: table1, ErrorMsg: err1},
+				},
+			},
+			table: table1,
+
+			wantTableErr: true,
+		},
+		{
+			name: "wildcard table",
+			err: &Errors{
+				Tables: []TableError{
+					{Table: table1, ErrorMsg: err1},
+				},
+			},
+			table: "*",
+
+			wantTableErr: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			got := tc.err.IsTableError(tc.table)
+			require.Equal(t, tc.wantTableErr, got)
+		})
+	}
+}
+
+func TestErrors_GetFailedTables(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  *Errors
+
+		wantTables []string
+	}{
+		{
+			name: "nil error",
+			err:  nil,
+
+			wantTables: nil,
+		},
+		{
+			name: "no failed tables",
+			err:  &Errors{},
+
+			wantTables: []string{},
+		},
+		{
+			name: "with failed tables",
+			err: &Errors{
+				Tables: []TableError{
+					{Table: table1, ErrorMsg: err1},
+					{Table: table2, ErrorMsg: err2},
+				},
+			},
+
+			wantTables: []string{table1, table2},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tables := tc.err.GetFailedTables()
+			require.Equal(t, tc.wantTables, tables)
+		})
+	}
+}
+
+func TestErrors_IsSnapshotError(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  *Errors
+
+		wantIsSnapshotErr bool
+	}{
+		{
+			name: "nil error",
+			err:  nil,
+
+			wantIsSnapshotErr: false,
+		},
+		{
+			name: "with snapshot error",
+			err: &Errors{
+				SnapshotErrMsgs: []string{err1, err2},
+			},
+
+			wantIsSnapshotErr: true,
+		},
+		{
+			name: "with table error",
+			err: &Errors{
+				Tables: []TableError{
+					NewTableError(table1, errTest),
+				},
+			},
+
+			wantIsSnapshotErr: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			got := tc.err.IsSnapshotError()
+			require.Equal(t, tc.wantIsSnapshotErr, got)
+		})
+	}
+}
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder_test.go b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
index 1ec591e..5dd09f4 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder_test.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
@@ -100,7 +100,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 			generator: &mockGenerator{},
 
-			wantErr: &snapshot.Errors{Snapshot: fmt.Errorf("retrieving existing snapshots for schema: %w", errTest)},
+			wantErr: snapshot.NewErrors(fmt.Errorf("retrieving existing snapshots for schema: %w", errTest)),
 		},
 		{
 			name: "error - snapshot error on wrapped generator",
@@ -126,7 +126,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 						require.Equal(t, &snapshot.Request{
 							Snapshot: testSnapshot,
 							Status:   snapshot.StatusCompleted,
-							Errors:   &snapshot.Errors{Snapshot: errTest},
+							Errors:   snapshot.NewErrors(errTest),
 						}, r)
 						return nil
 					default:
@@ -137,11 +137,11 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 			generator: &mockGenerator{
 				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
-					return &snapshot.Errors{Snapshot: errTest}
+					return snapshot.NewErrors(errTest)
 				},
 			},
 
-			wantErr: &snapshot.Errors{Snapshot: errTest},
+			wantErr: snapshot.NewErrors(errTest),
 		},
 		{
 			name: "error - recording snapshot request",
@@ -159,7 +159,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{Snapshot: errTest},
+			wantErr: snapshot.NewErrors(errTest),
 		},
 		{
 			name: "error - updating snapshot request in progress",
@@ -186,7 +186,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 			generator: &mockGenerator{},
 
-			wantErr: &snapshot.Errors{Snapshot: errTest},
+			wantErr: snapshot.NewErrors(errTest),
 		},
 		{
 			name: "error - updating snapshot request completed without errors",
@@ -227,7 +227,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				},
 			},
 
-			wantErr: &snapshot.Errors{Snapshot: errTest},
+			wantErr: snapshot.NewErrors(errTest),
 		},
 		{
 			name: "error - updating snapshot request completed with snapshot and table errors",
@@ -254,7 +254,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 							Snapshot: testSnapshot,
 							Status:   snapshot.StatusCompleted,
 							Errors: &snapshot.Errors{
-								Snapshot: errTest,
+								SnapshotErrMsgs: []string{errTest.Error()},
 								Tables: []snapshot.TableError{
 									{Table: "table1", ErrorMsg: errTest.Error()},
 								},
@@ -271,7 +271,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
 					require.Equal(t, newTestSnapshot(), ss)
 					return &snapshot.Errors{
-						Snapshot: errTest,
+						SnapshotErrMsgs: []string{errTest.Error()},
 						Tables: []snapshot.TableError{
 							{Table: "table1", ErrorMsg: errTest.Error()},
 						},
@@ -280,7 +280,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 
 			wantErr: &snapshot.Errors{
-				Snapshot: errors.Join(errTest, updateErr),
+				SnapshotErrMsgs: []string{errTest.Error(), updateErr.Error()},
 				Tables: []snapshot.TableError{
 					{Table: "table1", ErrorMsg: errTest.Error()},
 				},
@@ -335,7 +335,7 @@ func TestSnapshotRecorder_CreateSnapshot(t *testing.T) {
 			},
 
 			wantErr: &snapshot.Errors{
-				Snapshot: updateErr,
+				SnapshotErrMsgs: []string{updateErr.Error()},
 				Tables: []snapshot.TableError{
 					{Table: "table1", ErrorMsg: errTest.Error()},
 				},
@@ -461,7 +461,7 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 								SchemaName: testSchema,
 								TableNames: []string{"table2"},
 							},
-							Errors: &snapshot.Errors{Snapshot: errTest},
+							Errors: snapshot.NewErrors(errTest),
 							Status: snapshot.StatusCompleted,
 						},
 					}, nil
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
index 25db8b0..808d66b 100644
--- a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
@@ -1,4 +1,4 @@
-// SPDX-License-Identifier: Apache-2.0
+// // SPDX-License-Identifier: Apache-2.0
 
 package postgres
 
@@ -88,7 +88,7 @@ func TestStore_UpdateSnapshotRequest(t *testing.T) {
 	}
 
 	errTest := errors.New("oh noes")
-	testSnapshotErr := &snapshot.Errors{Snapshot: errTest}
+	testSnapshotErr := snapshot.NewErrors(errTest)
 
 	tests := []struct {
 		name    string
@@ -171,7 +171,7 @@ func TestStore_GetSnapshotRequestsBySchema(t *testing.T) {
 	testSnapshotRequest := snapshot.Request{
 		Snapshot: testSnapshot,
 		Status:   snapshot.StatusInProgress,
-		Errors:   &snapshot.Errors{Snapshot: errTest},
+		Errors:   snapshot.NewErrors(errTest),
 	}
 
 	tests := []struct {
@@ -298,7 +298,7 @@ func TestStore_GetSnapshotRequestsByStatus(t *testing.T) {
 	testSnapshotRequest := snapshot.Request{
 		Snapshot: testSnapshot,
 		Status:   snapshot.StatusInProgress,
-		Errors:   &snapshot.Errors{Snapshot: errTest},
+		Errors:   snapshot.NewErrors(errTest),
 	}
 
 	tests := []struct {