diff --git a/dbkit/CHANGELOG.md b/dbkit/CHANGELOG.md index 48e03f9..00fc1a6 100644 --- a/dbkit/CHANGELOG.md +++ b/dbkit/CHANGELOG.md @@ -8,6 +8,12 @@ v0.x, minor releases may contain breaking changes. Releases of this module are tagged `dbkit/vX.Y.Z`. +## [Unreleased] + +### Added + +- `InTx`, which runs a function in one write transaction on a plain `*sql.DB`, rolls back when it does not return nil, commits otherwise and joins a failed rollback to its error. + ## [0.2.0] - 2026-10-10 ### Added diff --git a/dbkit/fakedriver_test.go b/dbkit/fakedriver_test.go index 6b1580b..c48885a 100644 --- a/dbkit/fakedriver_test.go +++ b/dbkit/fakedriver_test.go @@ -122,9 +122,9 @@ func (f *fakeDriver) holdRollback() chan struct{} { return f.holdEnd(&f.rollback) } -// holdCommit makes every commit wait, once reported, until the returned gate closes or its begin context ends. -func (f *fakeDriver) holdCommit() chan struct{} { - return f.holdEnd(&f.commit) +// holdCommit makes every commit wait, once reported, until its begin context ends. +func (f *fakeDriver) holdCommit() { + f.holdEnd(&f.commit) } // holdEnd sets the gate at slot and returns it. diff --git a/dbkit/intx.go b/dbkit/intx.go new file mode 100644 index 0000000..04c2207 --- /dev/null +++ b/dbkit/intx.go @@ -0,0 +1,33 @@ +// SPDX-License-Identifier: Apache-2.0 + +package dbkit + +import ( + "context" + "database/sql" + "errors" +) + +// InTx runs body in a write transaction on db, rolling back when body does not return nil and committing otherwise. +func InTx(ctx context.Context, db *sql.DB, body func(*sql.Tx) error) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if err := body(tx); err != nil { + return rollbackAfter(err, tx) + } + if err := ended(ctx, tx.Commit()); err != nil { + return rollbackAfter(err, tx) + } + return nil +} + +// rollbackAfter rolls tx back and returns err, joined with the rollback error when the rollback fails. +func rollbackAfter(err error, tx *sql.Tx) error { + if rollbackErr := tx.Rollback(); rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) { + return errors.Join(err, rollbackErr) + } + return err +} diff --git a/dbkit/intx_test.go b/dbkit/intx_test.go new file mode 100644 index 0000000..1df304c --- /dev/null +++ b/dbkit/intx_test.go @@ -0,0 +1,175 @@ +// SPDX-License-Identifier: Apache-2.0 + +package dbkit_test + +import ( + "context" + "database/sql" + "errors" + "runtime" + "slices" + "testing" + "testing/synctest" + "time" + + "github.com/gopherium/framework/dbkit" +) + +// intxTimeout is the deadline of the InTx tests whose driver holds a call until the context ends. +const intxTimeout = time.Second + +// errBody is what a failing InTx body returns. +var errBody = errors.New("body failed") + +// wantCalls fails t unless the fake driver received exactly ops, in order. +func wantCalls(t *testing.T, fake *fakeDriver, ops ...string) { + t.Helper() + if got := fake.log(); !slices.Equal(got, ops) { + t.Errorf("driver calls = %v, want %v", got, ops) + } +} + +func TestInTxCommitsWhenBodySucceeds(t *testing.T) { + t.Parallel() + db, fake := shareHandle(t) + + err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { return nil }) + + if err != nil { + t.Errorf("InTx() error = %v, want nil", err) + } + wantCalls(t, fake, "begin", "commit") +} + +func TestInTxRollsBackWhenBodyReturnsAnError(t *testing.T) { + t.Parallel() + db, fake := shareHandle(t) + + err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { return errBody }) + + if !errors.Is(err, errBody) { + t.Errorf("InTx() error = %v, want %v", err, errBody) + } + wantCalls(t, fake, "begin", "rollback") +} + +func TestInTxRollsBackBeforeRethrowingAPanic(t *testing.T) { + t.Parallel() + db, fake := shareHandle(t) + want := "body panicked" + + defer func() { + if got := recover(); got != want { + t.Errorf("panic = %v, want %q", got, want) + } + wantCalls(t, fake, "begin", "rollback") + }() + _ = dbkit.InTx(t.Context(), db, func(*sql.Tx) error { panic(want) }) +} + +func TestInTxRollsBackWhenBodyNeverReturns(t *testing.T) { + t.Parallel() + db, fake := shareHandle(t) + done := make(chan struct{}) + + go func() { + defer close(done) + _ = dbkit.InTx(t.Context(), db, func(*sql.Tx) error { + runtime.Goexit() + return nil + }) + }() + <-done + + wantCalls(t, fake, "begin", "rollback") +} + +func TestInTxAnswersTheEndOfItsContext(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + db, fake := shareHandle(t) + ctx, cancel := context.WithCancel(t.Context()) + + err := dbkit.InTx(ctx, db, func(*sql.Tx) error { + cancel() + synctest.Wait() + return nil + }) + + if !errors.Is(err, context.Canceled) { + t.Errorf("InTx() error = %v, want it to match context.Canceled", err) + } + wantCalls(t, fake, "begin", "rollback") + }) +} + +func TestInTxAnswersABeginFailure(t *testing.T) { + t.Parallel() + db, fake := shareHandle(t) + fake.fail("") + ran := false + + err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { + ran = true + return nil + }) + + if !errors.Is(err, fakeBroken) { + t.Errorf("InTx() error = %v, want %v", err, fakeBroken) + } + if ran { + t.Error("the body ran, want it never to run once the begin failed") + } +} + +func TestInTxAnswersACommitFailure(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + db, fake := shareHandle(t) + fake.holdCommit() + ctx, cancel := context.WithTimeout(t.Context(), intxTimeout) + defer cancel() + + err := dbkit.InTx(ctx, db, func(*sql.Tx) error { return nil }) + + wantFailedCommit(t, err) + wantCalls(t, fake, "begin", "commit") + }) +} + +// wantFailedCommit fails t unless err is the driver's commit error marked with the deadline, never sql.ErrTxDone. +func wantFailedCommit(t *testing.T, err error) { + t.Helper() + if !errors.Is(err, fakeInterrupted) { + t.Errorf("InTx() error = %v, want the driver error %v", err, fakeInterrupted) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("InTx() error = %v, want it to match context.DeadlineExceeded", err) + } + if errors.Is(err, sql.ErrTxDone) { + t.Errorf("InTx() error = %v, want no %v joined", err, sql.ErrTxDone) + } +} + +func TestInTxJoinsAFailedRollbackToTheBodysError(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + db, fake := shareHandle(t) + fake.holdRollback() + ctx, cancel := context.WithTimeout(t.Context(), intxTimeout) + defer cancel() + + err := dbkit.InTx(ctx, db, func(*sql.Tx) error { return errBody }) + + if !errors.Is(err, errBody) { + t.Errorf("InTx() error = %v, want the body's %v", err, errBody) + } + if !errors.Is(err, fakeInterrupted) { + t.Errorf("InTx() error = %v, want the rollback's %v joined", err, fakeInterrupted) + } + wantCalls(t, fake, "begin", "rollback") + }) +}