diff --git a/internal/dumper/dump_errors_test.go b/internal/dumper/dump_errors_test.go new file mode 100644 index 000000000..d4c238be9 --- /dev/null +++ b/internal/dumper/dump_errors_test.go @@ -0,0 +1,273 @@ +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 + rowValuesErr error + closeCalled bool +} + +func (r *errCloseRows) Next() bool { + if r.rowSent { + return false + } + r.rowSent = true + return true +} + +func (r *errCloseRows) Close() error { + r.closeCalled = true + 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) { + if r.rowValuesErr != nil { + return nil, r.rowValuesErr + } + 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 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) + + 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..0aea69eac 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 @@ -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,24 +297,37 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin if err != nil { return err } - defer cursor.Close() + // 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 + } + }() 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 } @@ -324,7 +337,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 } @@ -342,7 +356,15 @@ func (d *Dumper) dumpTable(ctx context.Context, conn *Connection, database strin } } - if err := writer.Close(d.cfg.Outdir, database, table, fileNo); err != nil { + if err = writer.Close(d.cfg.Outdir, database, table, fileNo); err != nil { + return err + } + + // Close before the success log so a stream/network error is not preceded by "done". + cerr := cursor.Close() + closed = true + if cerr != nil { + err = cerr return err }