feat(dbkit): add transaction runner

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

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.

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
+}

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")
+	})
+}