From bdce942520010b499b1125d409a12aa66603c493 Mon Sep 17 00:00:00 2001 From: KeuqS Date: Wed, 14 Oct 2026 12:37:22 +0300 Subject: [PATCH 1/5] feat(dbkit): add transaction runner --- dbkit/tx.go | 34 ++++++++++++++++++++++++++++++++++ dbkit/tx_test.go | 43 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/dbkit/tx.go b/dbkit/tx.go index 86e3c36..c717249 100644 --- a/dbkit/tx.go +++ b/dbkit/tx.go @@ -5,11 +5,45 @@ package dbkit import ( "context" "database/sql" + "errors" "fmt" "sync" "time" ) +// InTx runs body in a transaction, committing it when body succeeds and +// rolling it back when body returns an error or panics. +func InTx(ctx context.Context, db *sql.DB, body func(*sql.Tx) error) (err error) { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { + if recovered := recover(); recovered != nil { + _ = tx.Rollback() + panic(recovered) + } + if err != nil { + err = rollbackAfter(err, tx) + return + } + if err = tx.Commit(); err != nil { + err = rollbackAfter(err, tx) + } + }() + + err = body(tx) + return err +} + +// rollbackAfter preserves the original error unless rollback fails independently. +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 +} + // Tx is a transaction of a share that holds one slot until Commit, Rollback or the end of its context. type Tx struct { share *Share diff --git a/dbkit/tx_test.go b/dbkit/tx_test.go index 8bc6850..31715e9 100644 --- a/dbkit/tx_test.go +++ b/dbkit/tx_test.go @@ -365,3 +365,46 @@ func TestABeginErrorComesBackAsItIs(t *testing.T) { shareFree(t, share, fake) }) } + +func TestInTxCommitsWhenBodySucceeds(t *testing.T) { + db, fake := shareHandle(t) + + if err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { return nil }); err != nil { + t.Fatalf("InTx() error = %v, want nil", err) + } + if !fake.saw("begin", "") || !fake.saw("commit", "") { + t.Fatalf("driver calls = %v, want begin and commit", fake.log()) + } + if fake.saw("rollback", "") { + t.Fatalf("driver calls = %v, want no rollback", fake.log()) + } +} + +func TestInTxRollsBackWhenBodyReturnsAnError(t *testing.T) { + db, fake := shareHandle(t) + want := errors.New("body failed") + + if err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { return want }); !errors.Is(err, want) { + t.Fatalf("InTx() error = %v, want %v", err, want) + } + if !fake.saw("rollback", "") || fake.saw("commit", "") { + t.Fatalf("driver calls = %v, want rollback and no commit", fake.log()) + } +} + +func TestInTxRollsBackBeforeRethrowingAPanic(t *testing.T) { + db, fake := shareHandle(t) + want := "body panicked" + + defer func() { + if got := recover(); got != want { + t.Fatalf("panic = %v, want %q", got, want) + } + if !fake.saw("rollback", "") || fake.saw("commit", "") { + t.Fatalf("driver calls = %v, want rollback and no commit", fake.log()) + } + }() + dbkit.InTx(t.Context(), db, func(*sql.Tx) error { + panic(want) + }) +} From 041da9ca4c9208d81666672aeaff0cd7e7c12bc1 Mon Sep 17 00:00:00 2001 From: KeuqS Date: Wed, 14 Oct 2026 12:41:03 +0300 Subject: [PATCH 2/5] test(dbkit): check transaction runner error --- dbkit/tx_test.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/dbkit/tx_test.go b/dbkit/tx_test.go index 31715e9..62e0af9 100644 --- a/dbkit/tx_test.go +++ b/dbkit/tx_test.go @@ -404,7 +404,9 @@ func TestInTxRollsBackBeforeRethrowingAPanic(t *testing.T) { t.Fatalf("driver calls = %v, want rollback and no commit", fake.log()) } }() - dbkit.InTx(t.Context(), db, func(*sql.Tx) error { + if err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { panic(want) - }) + }); err != nil { + t.Fatalf("InTx() error = %v, want panic", err) + } } From b0343b3d547a47725db5517168878eeaee5b8ef6 Mon Sep 17 00:00:00 2001 From: SirLouen Date: Thu, 15 Oct 2026 11:43:00 +0200 Subject: [PATCH 3/5] fix(dbkit): roll back an InTx whose body never returned --- dbkit/fakedriver_test.go | 6 +- dbkit/intx.go | 33 ++++++++ dbkit/intx_test.go | 175 +++++++++++++++++++++++++++++++++++++++ dbkit/tx.go | 34 -------- dbkit/tx_test.go | 45 ---------- 5 files changed, 211 insertions(+), 82 deletions(-) create mode 100644 dbkit/intx.go create mode 100644 dbkit/intx_test.go 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..9708bfd --- /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, committing only when body returns nil and rolling back 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") + }) +} diff --git a/dbkit/tx.go b/dbkit/tx.go index c717249..86e3c36 100644 --- a/dbkit/tx.go +++ b/dbkit/tx.go @@ -5,45 +5,11 @@ package dbkit import ( "context" "database/sql" - "errors" "fmt" "sync" "time" ) -// InTx runs body in a transaction, committing it when body succeeds and -// rolling it back when body returns an error or panics. -func InTx(ctx context.Context, db *sql.DB, body func(*sql.Tx) error) (err error) { - tx, err := db.BeginTx(ctx, nil) - if err != nil { - return err - } - defer func() { - if recovered := recover(); recovered != nil { - _ = tx.Rollback() - panic(recovered) - } - if err != nil { - err = rollbackAfter(err, tx) - return - } - if err = tx.Commit(); err != nil { - err = rollbackAfter(err, tx) - } - }() - - err = body(tx) - return err -} - -// rollbackAfter preserves the original error unless rollback fails independently. -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 -} - // Tx is a transaction of a share that holds one slot until Commit, Rollback or the end of its context. type Tx struct { share *Share diff --git a/dbkit/tx_test.go b/dbkit/tx_test.go index 62e0af9..8bc6850 100644 --- a/dbkit/tx_test.go +++ b/dbkit/tx_test.go @@ -365,48 +365,3 @@ func TestABeginErrorComesBackAsItIs(t *testing.T) { shareFree(t, share, fake) }) } - -func TestInTxCommitsWhenBodySucceeds(t *testing.T) { - db, fake := shareHandle(t) - - if err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { return nil }); err != nil { - t.Fatalf("InTx() error = %v, want nil", err) - } - if !fake.saw("begin", "") || !fake.saw("commit", "") { - t.Fatalf("driver calls = %v, want begin and commit", fake.log()) - } - if fake.saw("rollback", "") { - t.Fatalf("driver calls = %v, want no rollback", fake.log()) - } -} - -func TestInTxRollsBackWhenBodyReturnsAnError(t *testing.T) { - db, fake := shareHandle(t) - want := errors.New("body failed") - - if err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { return want }); !errors.Is(err, want) { - t.Fatalf("InTx() error = %v, want %v", err, want) - } - if !fake.saw("rollback", "") || fake.saw("commit", "") { - t.Fatalf("driver calls = %v, want rollback and no commit", fake.log()) - } -} - -func TestInTxRollsBackBeforeRethrowingAPanic(t *testing.T) { - db, fake := shareHandle(t) - want := "body panicked" - - defer func() { - if got := recover(); got != want { - t.Fatalf("panic = %v, want %q", got, want) - } - if !fake.saw("rollback", "") || fake.saw("commit", "") { - t.Fatalf("driver calls = %v, want rollback and no commit", fake.log()) - } - }() - if err := dbkit.InTx(t.Context(), db, func(*sql.Tx) error { - panic(want) - }); err != nil { - t.Fatalf("InTx() error = %v, want panic", err) - } -} From 8ae0aebb304022f5f99ac2fc08dcc07dd8f5d9a1 Mon Sep 17 00:00:00 2001 From: SirLouen Date: Thu, 15 Oct 2026 11:43:01 +0200 Subject: [PATCH 4/5] docs(dbkit): note InTx under Unreleased --- dbkit/CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/dbkit/CHANGELOG.md b/dbkit/CHANGELOG.md index 48e03f9..cafa15f 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`, commits only when it returns nil, rolls back otherwise and joins a failed rollback to its error. + ## [0.2.0] - 2026-10-10 ### Added From 577e87d326227596ffd45b74d516bf51b4c08d75 Mon Sep 17 00:00:00 2001 From: SirLouen Date: Thu, 15 Oct 2026 12:02:02 +0200 Subject: [PATCH 5/5] docs(dbkit): say InTx rolls back only when its body fails --- dbkit/CHANGELOG.md | 2 +- dbkit/intx.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/dbkit/CHANGELOG.md b/dbkit/CHANGELOG.md index cafa15f..00fc1a6 100644 --- a/dbkit/CHANGELOG.md +++ b/dbkit/CHANGELOG.md @@ -12,7 +12,7 @@ Releases of this module are tagged `dbkit/vX.Y.Z`. ### Added -- `InTx`, which runs a function in one write transaction on a plain `*sql.DB`, commits only when it returns nil, rolls back otherwise and joins a failed rollback to its error. +- `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 diff --git a/dbkit/intx.go b/dbkit/intx.go index 9708bfd..04c2207 100644 --- a/dbkit/intx.go +++ b/dbkit/intx.go @@ -8,7 +8,7 @@ import ( "errors" ) -// InTx runs body in a write transaction on db, committing only when body returns nil and rolling back otherwise. +// 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 {