From ca239351d2d224477acf8599b6e43a583d9e145b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 17 Aug 2026 16:36:07 +0000 Subject: [PATCH 1/5] Fix database dump silent failures on cursor close errors Commit 4d3dc85 replaced an explicit cursor.Close() check with defer cursor.Close(), which swallowed stream errors from go-mysqlstack that only surface when the cursor is closed after Next() stops. Also propagate dumpTable errors from errgroup workers so failed dumps return a non-zero exit code instead of logging and continuing. Co-authored-by: Gomez --- internal/dumper/dump_errors_test.go | 221 ++++++++++++++++++++++++++++ internal/dumper/dumper.go | 10 +- 2 files changed, 228 insertions(+), 3 deletions(-) create mode 100644 internal/dumper/dump_errors_test.go diff --git a/internal/dumper/dump_errors_test.go b/internal/dumper/dump_errors_test.go new file mode 100644 index 000000000..b4e7cf5f2 --- /dev/null +++ b/internal/dumper/dump_errors_test.go @@ -0,0 +1,221 @@ +package dumper + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/planetscale/cli/internal/cmdutil" + "github.com/xelabs/go-mysqlstack/driver" + querypb "github.com/xelabs/go-mysqlstack/sqlparser/depends/query" + "github.com/xelabs/go-mysqlstack/sqlparser/depends/sqltypes" + "github.com/xelabs/go-mysqlstack/xlog" + + qt "github.com/frankban/quicktest" +) + +type errCloseRows struct { + row []sqltypes.Value + rowSent bool + closeErr error +} + +func (r *errCloseRows) Next() bool { + if r.rowSent { + return false + } + r.rowSent = true + return true +} + +func (r *errCloseRows) Close() error { + return r.closeErr +} + +func (r *errCloseRows) Datas() []byte { return nil } + +func (r *errCloseRows) Bytes() int { return 0 } + +func (r *errCloseRows) RowsAffected() uint64 { return 0 } + +func (r *errCloseRows) LastInsertID() uint64 { return 0 } + +func (r *errCloseRows) LastError() error { return nil } + +func (r *errCloseRows) Fields() []*querypb.Field { return nil } + +func (r *errCloseRows) RowValues() ([]sqltypes.Value, error) { + return r.row, nil +} + +type fakeConn struct { + fieldsResult *sqltypes.Result + streamRows driver.Rows +} + +func (f *fakeConn) Ping() error { return nil } + +func (f *fakeConn) Quit() {} + +func (f *fakeConn) Close() error { return nil } + +func (f *fakeConn) Closed() bool { return false } + +func (f *fakeConn) Cleanup() {} + +func (f *fakeConn) NextPacket() ([]byte, error) { return nil, nil } + +func (f *fakeConn) ConnectionID() uint32 { return 0 } + +func (f *fakeConn) InitDB(string) error { return nil } + +func (f *fakeConn) Command(byte) error { return nil } + +func (f *fakeConn) Query(string) (driver.Rows, error) { + return f.streamRows, nil +} + +func (f *fakeConn) Exec(string) error { return nil } + +func (f *fakeConn) FetchAll(string, int) (*sqltypes.Result, error) { + return f.fieldsResult, nil +} + +func (f *fakeConn) FetchAllWithFunc(string, int, driver.Func) (*sqltypes.Result, error) { + return f.fieldsResult, nil +} + +func (f *fakeConn) ComStatementPrepare(string) (*driver.Statement, error) { + return nil, fmt.Errorf("not implemented") +} + +func TestDumpTableReportsCursorCloseError(t *testing.T) { + c := qt.New(t) + + closeErr := errors.New("unexpected EOF") + fieldsResult := &sqltypes.Result{ + Fields: []*querypb.Field{ + {Name: "Field", Type: querypb.Type_VARCHAR}, + {Name: "Type", Type: querypb.Type_VARCHAR}, + {Name: "Null", Type: querypb.Type_VARCHAR}, + {Name: "Key", Type: querypb.Type_VARCHAR}, + {Name: "Default", Type: querypb.Type_VARCHAR}, + {Name: "Extra", Type: querypb.Type_VARCHAR}, + }, + Rows: [][]sqltypes.Value{ + testRow("id", ""), + }, + } + + conn := &Connection{ + ID: 0, + client: &fakeConn{ + fieldsResult: fieldsResult, + streamRows: &errCloseRows{ + row: []sqltypes.Value{ + sqltypes.MakeTrusted(querypb.Type_INT32, []byte("1")), + }, + closeErr: closeErr, + }, + }, + } + + cfg := NewDefaultConfig() + cfg.Outdir = c.TempDir() + cfg.ChunksizeInMB = 128 + cfg.StmtSize = 1000000 + + d := &Dumper{ + cfg: cfg, + log: cmdutil.NewZapLogger(false), + } + + err := d.dumpTable(context.Background(), conn, "test", "t1") + c.Assert(err, qt.ErrorIs, closeErr) +} + +func TestRunReturnsDumpError(t *testing.T) { + c := qt.New(t) + + log := xlog.NewStdLog(xlog.Level(xlog.INFO)) + fakedbs := driver.NewTestHandler(log) + server, err := driver.MockMysqlServer(log, fakedbs) + c.Assert(err, qt.IsNil) + c.Cleanup(func() { server.Close() }) + + address := server.Addr() + + schemaResult := &sqltypes.Result{ + Fields: []*querypb.Field{ + {Name: "Table", Type: querypb.Type_VARCHAR}, + {Name: "Create Table", Type: querypb.Type_VARCHAR}, + }, + Rows: [][]sqltypes.Value{ + { + sqltypes.MakeTrusted(querypb.Type_VARCHAR, []byte("t1")), + sqltypes.MakeTrusted(querypb.Type_VARCHAR, + []byte("CREATE TABLE `t1` (`id` int) ENGINE=InnoDB")), + }, + }, + } + + tablesResult := &sqltypes.Result{ + Fields: []*querypb.Field{ + {Name: "Tables_in_test", Type: querypb.Type_VARCHAR}, + }, + Rows: [][]sqltypes.Value{ + {sqltypes.MakeTrusted(querypb.Type_VARCHAR, []byte("t1"))}, + }, + } + + viewsResult := &sqltypes.Result{ + Fields: []*querypb.Field{ + {Name: "TABLE_NAME", Type: querypb.Type_VARCHAR}, + }, + Rows: [][]sqltypes.Value{}, + } + + fieldsResult := &sqltypes.Result{ + Fields: []*querypb.Field{ + {Name: "Field", Type: querypb.Type_VARCHAR}, + {Name: "Type", Type: querypb.Type_VARCHAR}, + {Name: "Null", Type: querypb.Type_VARCHAR}, + {Name: "Key", Type: querypb.Type_VARCHAR}, + {Name: "Default", Type: querypb.Type_VARCHAR}, + {Name: "Extra", Type: querypb.Type_VARCHAR}, + }, + Rows: [][]sqltypes.Value{ + testRow("id", ""), + }, + } + + dumpErr := errors.New("closed network connection") + fakedbs.AddQueryPattern("use .*", &sqltypes.Result{}) + fakedbs.AddQueryPattern("show create table .*", schemaResult) + fakedbs.AddQueryPattern("show tables from .*", tablesResult) + fakedbs.AddQueryPattern("select table_name \n\t\t\t from information_schema.tables \n\t\t\t where table_schema like 'test' \n\t\t\t and table_type = 'view'\n\t\t\t", viewsResult) + fakedbs.AddQueryPattern("show fields from .*", fieldsResult) + fakedbs.AddQueryErrorPattern("select .* from `test`\\.`t1` .*", dumpErr) + fakedbs.AddQueryPattern("set .*", &sqltypes.Result{}) + + cfg := &Config{ + Database: "test", + Table: "t1", + Outdir: c.TempDir(), + User: "mock", + Password: "mock", + Address: address, + ChunksizeInMB: 1, + Threads: 1, + StmtSize: 10000, + IntervalMs: 500, + } + + d, err := NewDumper(cfg) + c.Assert(err, qt.IsNil) + + err = d.Run(context.Background()) + c.Assert(err, qt.IsNotNil) + c.Assert(err.Error(), qt.Contains, "closed network connection") +} diff --git a/internal/dumper/dumper.go b/internal/dumper/dumper.go index 45aeb7796..a053d8d76 100644 --- a/internal/dumper/dumper.go +++ b/internal/dumper/dumper.go @@ -198,9 +198,9 @@ func (d *Dumper) Run(ctx context.Context) error { zap.Int("thread_conn_id", conn.ID), ) - err := d.dumpTable(ctx, conn, database, table) - if err != nil { + if err := d.dumpTable(ctx, conn, database, table); err != nil { d.log.Error("error dumping table", zap.Error(err)) + return err } return nil @@ -297,7 +297,6 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin if err != nil { return err } - defer cursor.Close() var allBytes uint64 var allRows uint64 @@ -343,6 +342,11 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin } if err := writer.Close(d.cfg.Outdir, database, table, fileNo); err != nil { + _ = cursor.Close() + return err + } + + if err := cursor.Close(); err != nil { return err } From 0314cb6956131d7b13b69af54de104d1fd46a0ab Mon Sep 17 00:00:00 2001 From: Elom Gomez Date: Mon, 17 Aug 2026 12:30:54 -0500 Subject: [PATCH 2/5] Close dump cursors on early error returns. Keep Close error reporting via a named return defer so mid-loop failures still drain the stream before the connection returns to the pool. Co-authored-by: Cursor --- internal/dumper/dump_errors_test.go | 58 +++++++++++++++++++++++++++-- internal/dumper/dumper.go | 32 +++++++++------- 2 files changed, 74 insertions(+), 16 deletions(-) diff --git a/internal/dumper/dump_errors_test.go b/internal/dumper/dump_errors_test.go index b4e7cf5f2..8e90672dd 100644 --- a/internal/dumper/dump_errors_test.go +++ b/internal/dumper/dump_errors_test.go @@ -16,9 +16,11 @@ import ( ) type errCloseRows struct { - row []sqltypes.Value - rowSent bool - closeErr error + row []sqltypes.Value + rowSent bool + closeErr error + rowValuesErr error + closeCalled bool } func (r *errCloseRows) Next() bool { @@ -30,6 +32,7 @@ func (r *errCloseRows) Next() bool { } func (r *errCloseRows) Close() error { + r.closeCalled = true return r.closeErr } @@ -46,6 +49,9 @@ func (r *errCloseRows) LastError() error { return nil } func (r *errCloseRows) Fields() []*querypb.Field { return nil } func (r *errCloseRows) RowValues() ([]sqltypes.Value, error) { + if r.rowValuesErr != nil { + return nil, r.rowValuesErr + } return r.row, nil } @@ -135,6 +141,52 @@ func TestDumpTableReportsCursorCloseError(t *testing.T) { c.Assert(err, qt.ErrorIs, closeErr) } +func TestDumpTableClosesCursorOnRowValuesError(t *testing.T) { + c := qt.New(t) + + rowErr := errors.New("row decode failed") + fieldsResult := &sqltypes.Result{ + Fields: []*querypb.Field{ + {Name: "Field", Type: querypb.Type_VARCHAR}, + {Name: "Type", Type: querypb.Type_VARCHAR}, + {Name: "Null", Type: querypb.Type_VARCHAR}, + {Name: "Key", Type: querypb.Type_VARCHAR}, + {Name: "Default", Type: querypb.Type_VARCHAR}, + {Name: "Extra", Type: querypb.Type_VARCHAR}, + }, + Rows: [][]sqltypes.Value{ + testRow("id", ""), + }, + } + + rows := &errCloseRows{ + rowValuesErr: rowErr, + closeErr: errors.New("should not mask row error"), + } + + conn := &Connection{ + ID: 0, + client: &fakeConn{ + fieldsResult: fieldsResult, + streamRows: rows, + }, + } + + cfg := NewDefaultConfig() + cfg.Outdir = c.TempDir() + cfg.ChunksizeInMB = 128 + cfg.StmtSize = 1000000 + + d := &Dumper{ + cfg: cfg, + log: cmdutil.NewZapLogger(false), + } + + err := d.dumpTable(context.Background(), conn, "test", "t1") + c.Assert(err, qt.ErrorIs, rowErr) + c.Assert(rows.closeCalled, qt.IsTrue) +} + func TestRunReturnsDumpError(t *testing.T) { c := qt.New(t) diff --git a/internal/dumper/dumper.go b/internal/dumper/dumper.go index a053d8d76..166352600 100644 --- a/internal/dumper/dumper.go +++ b/internal/dumper/dumper.go @@ -272,7 +272,7 @@ func (d *Dumper) dumpTableSchema(conn *Connection, database string, table string } // Dump a table in the configured output format -func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database string, table string) error { +func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database string, table string) (err error) { var writer TableWriter switch d.cfg.OutputFormat { @@ -297,23 +297,33 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin if err != nil { return err } + // Always close the stream so pooled connections are not left mid-result-set. + // Preserve the primary error; only surface Close failures when the dump otherwise succeeded. + defer func() { + if cerr := cursor.Close(); cerr != nil && err == nil { + err = cerr + } + }() var allBytes uint64 var allRows uint64 fileNo := 1 for cursor.Next() { - row, err := cursor.RowValues() - if err != nil { + row, rowErr := cursor.RowValues() + if rowErr != nil { + err = rowErr return err } // Allows for quicker exit when using Ctrl+C at the Terminal: if ctx.Err() != nil { - return ctx.Err() + err = ctx.Err() + return err } - bytesAdded, err := writer.WriteRow(row) - if err != nil { + bytesAdded, writeErr := writer.WriteRow(row) + if writeErr != nil { + err = writeErr return err } @@ -323,7 +333,8 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin atomic.AddUint64(&d.cfg.Allrows, 1) if writer.ShouldFlush() { - if err := writer.Flush(d.cfg.Outdir, database, table, fileNo); err != nil { + if flushErr := writer.Flush(d.cfg.Outdir, database, table, fileNo); flushErr != nil { + err = flushErr return err } @@ -341,12 +352,7 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin } } - if err := writer.Close(d.cfg.Outdir, database, table, fileNo); err != nil { - _ = cursor.Close() - return err - } - - if err := cursor.Close(); err != nil { + if err = writer.Close(d.cfg.Outdir, database, table, fileNo); err != nil { return err } From 0032601c32fc2d66dd79b3b109b43a7fd3830cf4 Mon Sep 17 00:00:00 2001 From: Elom Gomez Date: Mon, 17 Aug 2026 12:35:03 -0500 Subject: [PATCH 3/5] Fix goimports alignment in dump error tests. Co-authored-by: Cursor --- internal/dumper/dump_errors_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/dumper/dump_errors_test.go b/internal/dumper/dump_errors_test.go index 8e90672dd..d4c238be9 100644 --- a/internal/dumper/dump_errors_test.go +++ b/internal/dumper/dump_errors_test.go @@ -16,11 +16,11 @@ import ( ) type errCloseRows struct { - row []sqltypes.Value - rowSent bool - closeErr error + row []sqltypes.Value + rowSent bool + closeErr error rowValuesErr error - closeCalled bool + closeCalled bool } func (r *errCloseRows) Next() bool { From c1443863d1f499a9f86e398e95c4a4f58f600948 Mon Sep 17 00:00:00 2001 From: Elom Gomez Date: Mon, 17 Aug 2026 12:39:30 -0500 Subject: [PATCH 4/5] Close dump cursor before logging table completion. Avoid a misleading "dumping table done" log when the deferred stream close is what fails the dump. Co-authored-by: Cursor --- internal/dumper/dumper.go | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/internal/dumper/dumper.go b/internal/dumper/dumper.go index 166352600..e2ec38708 100644 --- a/internal/dumper/dumper.go +++ b/internal/dumper/dumper.go @@ -299,7 +299,11 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin } // Always close the stream so pooled connections are not left mid-result-set. // Preserve the primary error; only surface Close failures when the dump otherwise succeeded. + closed := false defer func() { + if closed { + return + } if cerr := cursor.Close(); cerr != nil && err == nil { err = cerr } @@ -356,6 +360,12 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin return err } + // Close before the success log so a stream/network error is not preceded by "done". + if err = cursor.Close(); err != nil { + return err + } + closed = true + d.log.Info( "dumping table done...", zap.String("database", database), From 1db75b3c2e702b2deb25dc9e6218f8318f8672e9 Mon Sep 17 00:00:00 2001 From: Elom Gomez Date: Mon, 17 Aug 2026 12:44:12 -0500 Subject: [PATCH 5/5] Avoid double-closing dump cursors after an explicit close error. Co-authored-by: Cursor --- internal/dumper/dumper.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/internal/dumper/dumper.go b/internal/dumper/dumper.go index e2ec38708..0aea69eac 100644 --- a/internal/dumper/dumper.go +++ b/internal/dumper/dumper.go @@ -361,10 +361,12 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin } // Close before the success log so a stream/network error is not preceded by "done". - if err = cursor.Close(); err != nil { + cerr := cursor.Close() + closed = true + if cerr != nil { + err = cerr return err } - closed = true d.log.Info( "dumping table done...",