txnproof

package module
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 5, 2026 License: MIT Imports: 16 Imported by: 0

README

go-txnproof

.github/workflows/check.yml

Detects non-atomic SQL execution in Go applications: multiple write statements that run inside one logical boundary (a use case, a request, a job) without being wrapped in a single database transaction.

A crash or error between such writes leaves partial state behind. go-txnproof finds those spots — in unit tests, in tests against a real database, and continuously in production — with a single core mechanism.

How it works

go-txnproof is a database/sql driver middleware. It observes every statement, tracks driver-level Begin/Commit/Rollback to know whether each statement ran inside a transaction, and attributes statements to a logical boundary carried on context.Context.

At the end of a boundary it counts atomic units that contained writes:

  • every transaction that contained at least one write = 1 unit
  • every auto-commit write = 1 unit each

If a boundary's writes span 2 or more units, the boundary is not atomic and a Violation is reported to your configured Reporters.

Reads never count. Statements executed outside any boundary are ignored by default (see unbounded writes).

Install

go get github.com/moznion/go-txnproof

Zero dependencies outside the standard library.

Quick start

1. Wrap your driver
import (
	"database/sql"

	"github.com/jackc/pgx/v4/stdlib"
	"github.com/moznion/go-txnproof"
)

detector := txnproof.New(
	txnproof.WithReporter(txnproof.NewSlogReporter(nil)), // or your own Reporter
)

sql.Register("pgx-txnproof", detector.Wrap(stdlib.GetDefaultDriver()))
db, err := sql.Open("pgx-txnproof", dsn)

Any database/sql driver works (pgx, lib/pq, go-sql-driver/mysql, mattn/go-sqlite3, ...). If you use sql.OpenDB, wrap the connector instead with detector.WrapConnector.

Native drivers (pgx without database/sql)

A connection that never goes through database/sql has no driver to wrap. For those, feed statement text to a per-connection Session — the same state machine the driver middleware is built on. With pgx, a QueryTracer sees every statement including the textual begin/commit/rollback that Begin()/Commit() execute, so transaction attribution works from the text alone:

type tracer struct {
	det      *txnproof.Detector
	mu       sync.Mutex
	sessions map[*pgx.Conn]*txnproof.Session
}

func (t *tracer) TraceQueryStart(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
	t.sessionFor(conn).Observe(ctx, data.SQL) // one Session per connection
	return ctx
}

Install it via pgxpool.Config.ConnConfig.Tracer, and drop the connection's Session in pgxpool.Config.BeforeClose. Two details matter beyond the happy path: a batch (SendBatch) is pipelined up to a single Sync, so PostgreSQL runs it as one implicit transaction — bracket it with Session.BeginTx/EndTx when (and only when) the connection is idle at batch start; and a Session must be used serially per connection, which pgx already guarantees. The full reference implementation, cross-checked against the server's own log, is e2e/pgxtracer.go with its scenarios in e2e/e2e_pgx_test.go.

2. Mark boundaries

A boundary is the unit that should be atomic — typically a use case invocation. Put it in middleware so every code path is covered:

ctx, b := detector.StartBoundary(ctx, "CreateUser")
defer b.Finish()

// ... run the use case with ctx ...

or use the closure form:

err := detector.InBoundary(ctx, "CreateUser", func(ctx context.Context) error {
	return createUserUseCase.Do(ctx, input)
})

Every statement executed with that context (through the wrapped driver) is attributed to the boundary. When b.Finish() runs, violations are reported.

3. Testing without a database

NewNullDB returns a *sql.DB backed by an in-memory no-op driver: every statement succeeds and returns no rows, and only the statement/transaction timeline is observed. Unlike sqlmock, no expectations need to be declared — inject it and assert atomicity:

func TestCreateUserIsAtomic(t *testing.T) {
	reporter := txnproof.NewCollectingReporter()
	detector := txnproof.New(txnproof.WithReporter(reporter))
	db := detector.NewNullDB()

	uc := NewCreateUserUseCase(db)
	_ = detector.InBoundary(context.Background(), "CreateUser", func(ctx context.Context) error {
		return uc.Do(ctx, input)
	})

	reporter.RequireNoViolations(t)
}

For tests against a real database, keep your real driver and just wrap it (step 1) — the same assertions work, and the database actually executes the statements.

4. Production monitoring

The same wiring, with a monitoring reporter instead of a test reporter:

detector := txnproof.New(
	txnproof.WithReporter(txnproof.NewSlogReporter(logger)),
	// or ReporterFunc to emit metrics / notify an error tracker:
	txnproof.WithReporter(txnproof.ReporterFunc(func(ctx context.Context, v txnproof.Violation) {
		metrics.Count("txnproof.violation", 1, "boundary:"+v.Boundary)
	})),
)

The overhead is per-statement bookkeeping — a classification and a write-unit tally — with no extra I/O, and it is allocation-free per statement. See Performance.

Throttling reports on hot paths

A violating boundary on a hot path fires the reporter on every request. Wrap your monitoring reporter in a ThrottlingReporter to report each boundary at most once per interval:

throttled := txnproof.NewThrottlingReporter(txnproof.NewSlogReporter(logger), 10*time.Minute)
detector := txnproof.New(txnproof.WithReporter(throttled))

Per boundary name, the first violation is forwarded immediately; further violations for the same boundary within the interval are suppressed; after the interval elapses, the next one is forwarded again. The optional signals pass through with the same interval but independent windows: unbounded writes are deduplicated per statement text, stale AllowNonAtomic marks per boundary. Wrapping does not swallow these signals — they are forwarded whenever the wrapped reporter implements the corresponding interface.

Suppressed reports are counted, not lost. The cumulative counts are available as snapshots to log or export periodically:

go func() {
	for range time.Tick(time.Minute) {
		for boundary, n := range throttled.SuppressedViolations() {
			metrics.Gauge("txnproof.suppressed_violations", n, "boundary:"+boundary)
		}
	}
}()

Memory stays bounded: the boundary-keyed state grows only with the set of boundary names (code-defined and small), and the statement-keyed state is capped, beyond which new statements are reported unthrottled.

Allowing intentional non-atomicity

Some boundaries are intentionally non-atomic (best-effort audit writes, writes spanning two databases that a single transaction cannot cover). Suppressing them explicitly always requires a reason; what you choose is where the exemption lives — on the boundary, at the write that makes it non-atomic, or in a central list.

In-code, on the boundary: AllowNonAtomic

Mark the boundary at its call site — the reason lives next to the code, survives refactors, and shows up in code review diffs:

ctx, b := detector.StartBoundary(ctx, "WriteAuditLog",
	txnproof.AllowNonAtomic("audit writes are best-effort by design (TICKET-123)"))
defer b.Finish()

By default the mark allows any amount of non-atomicity in that boundary. Pin it to the exact write-unit count you reviewed to keep the exemption from silently growing — a boundary allowed at 2 units that later grows to 3 is an unreviewed violation again, and is reported as one:

ctx, b := detector.StartBoundary(ctx, "WriteAuditLog",
	// exactly the domain write plus the audit write, nothing more
	txnproof.AllowNonAtomic("audit writes are best-effort by design (TICKET-123)", 2))
defer b.Finish()

Both directions are caught, and both say what happened:

  • more (or fewer) units than declared, still non-atomic — reported as a Violation carrying AllowedWriteUnits, so the message reads writes span 3 atomic units (allowed for exactly 2 write unit(s), so this execution is not covered) instead of looking like an unmarked boundary.
  • declared non-atomic but actually atomic (fewer than 2 units) — no violation, because the execution was atomic; it surfaces through the stale-allow channel below, which fails RequireNoStaleAllows in tests.

Pass several counts (AllowNonAtomic(reason, 2, 3)) when the write count legitimately differs per code path; each listed count is allowed, everything else violates. Note that the number is Violation.WriteUnits, not the number of transactions: one transaction that wrote counts 1, and so does each auto-commit write. A count below 2 can never match a violation, so such a mark is stale on every execution (see below). When the count does not match, the boundary is treated as unmarked — the central Allowlist is still consulted afterwards. The same optional counts exist on Allowlist.Add, and both decide identically.

Rot prevention works per execution: when an allowed boundary finishes with fewer than 2 write units (the allow suppressed nothing), reporters implementing StaleAllowReporter (both CollectingReporter and SlogReporter do) are notified. In tests, assert it:

reporter.RequireNoViolations(t)
reporter.RequireNoStaleAllows(t)

Because a boundary's write count can vary by code path, a stale-allow report in production is a hint, not proof — an allowed boundary may violate on one request and not on the next. In deterministic tests it is exact.

In-code, at the write site: AllowNonAtomicHere

The boundary usually starts far from the code that makes it non-atomic — in a middleware or a use-case entry point, while the reason for the extra write is at the extra write. AllowNonAtomicHere marks the boundary in the context from there, so the explanation sits next to the code it explains and stays running code rather than a comment:

// The audit row is written outside the domain transaction on purpose, so a
// failing audit sink cannot roll back the business change (TICKET-123).
txnproof.AllowNonAtomicHere(ctx, "audit write is best-effort (TICKET-123)", 2)
if _, err := db.ExecContext(ctx, "INSERT INTO audit ..."); err != nil {
	return err
}

It is the same mark as the AllowNonAtomic option — same reason, same optional exact write-unit counts, same fall-through to the Allowlist, same stale-allow rot prevention — only declared elsewhere. Details:

  • The mark applies to the innermost boundary in ctx, consistent with how statements attribute, and the moment it is called does not matter: evaluation happens at Finish.
  • The last mark wins, replacing an earlier one — including one made by the AllowNonAtomic option, so a call site can narrow (or widen) what the boundary declared.
  • With no boundary in ctx, or after the boundary has finished, it does nothing — just as statements executed outside any boundary are ignored. Missing boundary plumbing is caught by WithUnboundedWriteDetection, which reports the write this call precedes.
  • Marking at the write site is usually the more durable of the two against rot: a mark on a conditional path exists only on the executions that reach it, whereas one declared at boundary start is stale on every execution that stays atomic.
Central list: Allowlist

Alternatively, keep exemptions in one place — convenient for bulk initial adoption on an existing codebase:

allowlist := txnproof.NewAllowlist().
	Add("WriteAuditLog", "audit writes are best-effort by design (TICKET-123)").
	Add("SyncToAnalyticsDB", "spans two databases; compensated by nightly reconcile (TICKET-456)")

detector := txnproof.New(
	txnproof.WithReporter(reporter),
	txnproof.WithAllowlist(allowlist),
)

Entries take the same optional exact write-unit counts as AllowNonAtomic, with the same meaning — the choice between the two mechanisms stays a question of where the exemption lives, never of what it can express:

allowlist := txnproof.NewAllowlist().
	// exactly the domain write plus the audit write, nothing more
	Add("WriteAuditLog", "audit writes are best-effort by design (TICKET-123)", 2)

To keep the list from rotting, every entry tracks whether it actually suppressed a violation. Fail CI when entries go stale — the same discipline as unused //nolint directives:

if unused := allowlist.UnusedEntries(); len(unused) > 0 {
	t.Errorf("stale allowlist entries (remove them): %v", unused)
}

An entry constrained to exact counts that stops matching also shows up as unused — together with the Violation for the uncovered count. That pair means the boundary changed, so review it (and the count) rather than deleting the entry outright.

The mechanisms coexist: the in-code mark wins first (whether made by AllowNonAtomic or AllowNonAtomicHere), then the Allowlist is consulted. A practical migration is to allowlist everything on first adoption, then move the permanent exemptions into the code.

Baseline / ratchet

Adopting txnproof on an existing codebase usually surfaces violations you cannot fix on day one. A baseline captures them once; from then on only new violations fail, and existing ones are tolerated until fixed — the same ratchet idea as golangci-lint --new-from-rev or rubocop --auto-gen-config.

Generate the baseline deliberately, once (e.g. behind an env var or a small helper command):

reporter := txnproof.NewCollectingReporter()
detector := txnproof.New(txnproof.WithReporter(reporter))
// ... run the full test suite / scenario ...

// Explicit, intentional write — txnproof never rewrites the file on its own.
err := txnproof.BaselineFromViolations(reporter.Violations()).Save("txnproof-baseline.json")

The file is deterministic, sorted, indented JSON keyed on boundary names (the stable identifier — no counts, statements, or timestamps), so diffs stay clean. Commit it.

On every subsequent run, load it and wrap your reporter — baselined boundaries are filtered out before the reporter sees them:

baseline, err := txnproof.LoadBaseline("txnproof-baseline.json")
if err != nil {
	// A missing file is an error on purpose: creating the baseline is a
	// deliberate Save call, never a silent fallback.
	log.Fatal(err)
}

reporter := txnproof.NewCollectingReporter()
detector := txnproof.New(
	txnproof.WithReporter(txnproof.NewBaselineReporter(baseline, reporter)),
)

// ... run the suite ...
reporter.RequireNoViolations(t) // fails only on violations NOT in the baseline

The ratchet must only go down: like Allowlist, every baseline entry tracks whether it actually suppressed a violation. Fail CI when a boundary got fixed but its entry lingers, and remove the entry (or regenerate the file intentionally):

if stale := baseline.UnusedEntries(); len(stale) > 0 {
	t.Errorf("boundaries fixed but still baselined (remove them from txnproof-baseline.json): %v", stale)
}

Baseline vs. Allowlist: an allowlist entry says "this is intentionally non-atomic, forever, for this reason"; a baseline entry says "this is a known bug we have not fixed yet". Debt goes in the baseline, design decisions go in the allowlist (or in-code AllowNonAtomic).

Unbounded writes

Writes executed with a context that carries no boundary (e.g. goroutines detached via context.Background()) cannot be attributed to any boundary — they never count toward any boundary's write units and never produce a Violation. A boundary that does one write itself and detaches a goroutine for a second write therefore looks atomic to txnproof: that is the blind spot this option surfaces. Opt in:

detector := txnproof.New(
	txnproof.WithReporter(reporter),
	txnproof.WithUnboundedWriteDetection(),
)

Reporters that implement UnboundedWriteReporter (both CollectingReporter and SlogReporter do) receive each unbounded write as it executes: one report per statement, immediately, with no threshold — a single write is reported. Reads are never reported. There is no judgment attached: an unbounded write may be a missing boundary, a context detached by mistake, or a deliberate background write — the report says only that a state change happened outside the detection net.

In tests, require full boundary coverage alongside atomicity:

reporter.RequireNoViolations(t)
reporter.RequireNoUnboundedWrites(t)

In production, "zero violations but unbounded writes present" is the signal to investigate boundary coverage. ThrottlingReporter deduplicates unbounded writes per statement text (see throttling), and WithBoundaryAttrsFunc is evaluated against each unbounded write's own context at record time (see tying violations back to requests), so even detached writes stay traceable.

Nested boundaries

Starting a boundary on a context that already carries one shadows the outer boundary: subsequent statements attribute to the inner boundary only. This is a deliberate contract, not an implementation limit — a boundary is the smallest unit that should be atomic, and the innermost declaration is the most specific claim. The alternative (counting every statement toward all enclosing boundaries) would flag every composite use case that calls two individually-atomic sub-use-cases, and whether such a composition must be atomic is a design decision (outbox/saga territory), not something a counter can rule on.

The trade-off: an outer boundary that writes directly and calls an inner boundary that also writes is non-atomic as a whole, yet each boundary sees only one write — no violation. And nesting itself is usually not intended at all: it typically means two instrumentation layers overlap (a per-request middleware and a per-use-case wrapper both starting boundaries). Opt in to make nesting observable:

detector := txnproof.New(
	txnproof.WithReporter(reporter),
	txnproof.WithNestedBoundaryDetection(),
)

Each nesting occurrence is delivered — immediately, at StartBoundary time, never as a Violation — to reporters implementing NestedBoundaryReporter (both CollectingReporter and SlogReporter do), carrying the outer and inner boundary names. In tests:

reporter.RequireNoNestedBoundaries(t)

ThrottlingReporter deduplicates nesting reports per outer/inner name pair, so an overlapping middleware stack on a hot path reports once per interval, not once per request.

Tying violations back to requests

A production Violation is only actionable if you can find the request that produced it. Boundary attrs are string-keyed values attached to a boundary and carried into every Violation it produces (Violation.Attrs); SlogReporter emits them as log attributes, and CollectingReporter exposes them on the stored violations.

Set up a detector-level extractor once — typically pulling trace/request IDs out of the context — and every boundary gets them for free. It runs once per boundary start, never per statement:

// TxnProofMiddleware opens a boundary per request, named after the matched
// route pattern (resolved with mux.Handler — the mux populates r.Pattern
// only after routing, so an outer middleware cannot read it). Use route
// patterns, never raw URL paths: boundary names key the allowlist, the
// baseline, and throttling state, so they must stay a small, code-defined
// set — "/users/123" would leak one boundary name per user.
func TxnProofMiddleware(detector *txnproof.Detector, mux *http.ServeMux) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, pattern := mux.Handler(r)
		if pattern == "" { // unmatched request (404)
			pattern = r.Method + " " + r.URL.Path
		}
		ctx, b := detector.StartBoundary(r.Context(), pattern)
		defer b.Finish()
		mux.ServeHTTP(w, r.WithContext(ctx))
	})
}

func main() {
	detector := txnproof.New(
		txnproof.WithReporter(txnproof.NewSlogReporter(logger)),
		txnproof.WithBoundaryAttrsFunc(func(ctx context.Context) []txnproof.BoundaryAttr {
			return []txnproof.BoundaryAttr{
				txnproof.Attr("request_id", requestid.FromContext(ctx)),
				txnproof.Attr("trace_id", trace.SpanContextFromContext(ctx).TraceID().String()),
			}
		}),
	)

	mux := http.NewServeMux()
	mux.HandleFunc("POST /users", createUserHandler)

	// The attrs func reads the context as it is when the boundary starts, so
	// TxnProofMiddleware goes INSIDE the middleware that puts the request and
	// trace IDs on the context.
	handler := requestid.Middleware(TxnProofMiddleware(detector, mux))
	log.Fatal(http.ListenAndServe(":8080", handler))
}

Values the caller already has at hand go on a single boundary with WithBoundaryAttrs:

ctx, b := detector.StartBoundary(ctx, "CreateUser",
	txnproof.WithBoundaryAttrs(txnproof.Attr("user_id", userID)))
defer b.Finish()

Both coexist: detector-level attrs come first, then per-boundary ones. Duplicate keys are kept in order, never deduplicated. Reporters built on log/slog can convert with txnproof.SlogAttrs.

With unbounded-write detection on, the extractor also runs for each unbounded write — at record time, against the statement's own context — and the result is delivered on StatementRecord.Attrs, so even detached writes stay traceable.

Cross-checking against server logs

txnproof's detection is a client-side observation, with documented blind spots (detached contexts, heuristic classification, best-effort textual BEGIN/COMMIT tracking). For tests that run against a real database, the crosscheck subpackage verifies atomicity from the server's own logs — the authoritative record of which transaction each statement actually ran in. crosscheck is database-agnostic: it groups the logged write statements by server-side transaction identity and applies the same semantics as txnproof (reads never count; a rolled-back transaction still counts as a unit). A database-specific adapter supplies the parsing; pgcheck is the PostgreSQL adapter, and writing one for another database means implementing a single interface (crosscheck.Parser) — see the crosscheck package documentation for the contract.

PostgreSQL: pgcheck

Configure the test database to log every statement with transaction identifiers, in the plain (stderr) log format with English message tags:

log_line_prefix = '%m [%p] %q%x %v '
log_statement = 'all'
lc_messages = 'C'

%v is the virtual transaction ID, assigned to every transaction including the implicit one of each auto-commit statement — so two auto-commit writes always show two different values, which is exactly what the cross-check catches, and log_statement logs each statement right after its transaction acquired one. %x (the real transaction ID) is a weaker fallback for prefixes without %v: reads never get one, and it reads 0 until a statement forces assignment — in particular, don't rely on log_min_duration_statement timing, because the duration line of an auto-commit statement is emitted after its implicit transaction already ended. A different log_line_prefix works via pgcheck.WithLogLinePrefix (translated automatically) as long as it contains %x and/or %v.

Delimit the scenario with marker statements and verify the log tail written during the test:

func TestCreateUserIsAtomicOnRealPostgres(t *testing.T) {
	dsn := os.Getenv("TXNPROOF_TEST_PG_DSN")
	if dsn == "" {
		t.Skip("TXNPROOF_TEST_PG_DSN not set")
	}
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	logFile := os.Getenv("TXNPROOF_TEST_PG_LOG") // the server's current stderr log file
	offset := fileSize(t, logFile)              // remember where the log ends before the scenario

	mustExec(t, db, pgcheck.BeginMarker("create-user"))
	runCreateUserUseCase(t, db) // the scenario under test
	mustExec(t, db, pgcheck.EndMarker("create-user"))

	checker, err := pgcheck.New()
	if err != nil {
		t.Fatal(err)
	}
	tail := openAt(t, logFile, offset) // read only what the scenario appended
	defer tail.Close()
	if _, err := checker.VerifyScenario(tail, "create-user"); err != nil {
		t.Fatal(err) // e.g. "crosscheck: writes span 2 server-side transactions (want 1) — ..."
	}
}

The markers make the check robust against unrelated log lines before and after the scenario; reading from a recorded offset keeps earlier test runs out. Statements from other connections interleaved inside the scenario would still be counted, so run such tests against a dedicated (or quiet) database. Both the simple and the extended query protocol (as used by pgx) are understood; multi-line statements are reassembled. Rolled-back transactions count as a unit, matching txnproof's client-side semantics.

MySQL: mycheck

MySQL puts no transaction identifier on log lines — there is nothing like PostgreSQL's %v for the parser to read. mycheck instead parses the general query log, which records every statement with its connection (thread) id, and reconstructs the transaction grouping per thread with a state machine over the statement stream: BEGIN/START TRANSACTION opens a transaction, COMMIT/ROLLBACK closes it (ROLLBACK TO SAVEPOINT does not), MySQL's implicit-commit statements (DDL and friends) close it, a connection that disconnects mid-transaction implicitly rolls back — the writes still count as a unit — and everything outside a transaction is its own auto-commit unit.

Be aware this is a weaker guarantee than pgcheck's: the grouping is inferred from the server's statement stream, not read from a server-assigned transaction id. It is still server-side truth about which statements actually ran, in what order, on which connection — exactly what catches detached-context writes and client-side classification misses — but a server behavior the state machine does not model (an unrecognized implicit-commit statement, a disabled autocommit mode) can mis-group. A SET touching autocommit inside the verified scenario is therefore a hard error rather than a guess. The binary log was rejected as the source: rolled-back transactions never reach it, which contradicts txnproof's core semantics.

Configure the test server to write the general query log to a file:

general_log = ON
log_output = 'FILE'
general_log_file = /path/to/general.log
log_timestamps = UTC

The general query log records every statement of every connection — use it in test environments, not production. Usage is the same as pgcheck: delimit the scenario with mycheck.BeginMarker / mycheck.EndMarker and run mycheck.New().VerifyScenario(tail, label) over the log tail. Both the plain query path and server-side prepared statements (logged as Execute entries with parameters substituted) are understood; multi-line statements are reassembled. The format was validated against MySQL 8.0, 8.4 (LTS), and 9.7; CI runs the e2e-mysql/ suite across all three release lines.

End-to-end self-verification

txnproof verifies itself with both lenses at once: the e2e/ and e2e-mysql/ modules (separate Go modules, excluded from the library's zero-dependency surface) run scenarios through a driver wrapped by txnproof against a real PostgreSQL / MySQL and require the client-side verdict and the server-log verdict (pgcheck / mycheck) to agree — including the tricky paths (rolled-back transactions, textual BEGIN/COMMIT, savepoints, the prepared-statement path). e2e/run.sh and e2e-mysql/run.sh spin up a throwaway server (no Docker needed) and run them; CI does the same on every push, across PostgreSQL 16–18.

Robustness

txnproof runs inside your application's request path as driver middleware, so a panic in it would take the application down with it — a far worse outcome than a missed violation. Every surface that consumes text txnproof does not control is therefore fuzzed on every change: the statement classifier, the throttle's key derivation, the baseline file loader, and the PostgreSQL / MySQL server-log parsers. The detector itself is fuzzed too — generated statement programs (boundaries, transactions, textual BEGIN/COMMIT, prepared statements) run against the null driver and every report is cross-checked against an independent model of the documented counting rules, so the fuzzer catches wrong verdicts and not just crashes.

Run a sweep yourself with make fuzz (or make fuzz FUZZTIME=5m for a longer one).

Performance

txnproof sits on the statement hot path, so it is built to stay out of the way. Its steady-state cost is zero allocations per statement and one allocation per boundary.

  • Per statement — 0 allocations. Classifying a statement and tallying its write-unit both run without touching the heap, regardless of the SQL's letter case (the classifier uppercases the leading keyword into a stack buffer rather than via strings.ToUpper). There is no extra I/O; the underlying driver does the same work it always did.
  • Prepared statements classify once. The statement kind is computed at Prepare and reused for every execution, so drivers with statement caching (e.g. pgx's database/sql layer) never re-pay classification per execution — this matters for data-modifying CTEs, whose classification scans the whole statement text.
  • Per boundary — 1 allocation. The Boundary struct is the only unavoidable allocation. It is returned as a context.Context and mutated from driver goroutines through ctx.Value, so it provably escapes to the heap — a structural floor, not an oversight. Reaching zero would require sync.Pool recycling, which is deliberately avoided: a context can outlive Finish (e.g. a goroutine that captured it and runs a query afterward), and a recycled boundary would then be mutated on behalf of a stale context — cross-boundary contamination, unacceptable for a correctness tool.
  • Opt-in / rare paths cost more, by design. The statement-record buffer for violation reports is allocated lazily on first use and bounded by WithMaxRecordedStatements; a boundary that spans more than four distinct write transactions spills its write-tx set into a small map. Neither is on the common, healthy path.

Measured on an Apple M4 Pro (go test -bench . -benchmem):

Path ns/op B/op allocs/op
Classify INSERT (lowercase) 10 0 0
Classify data-modifying CTE (WITH … DELETE … INSERT) 41 0 0
Classify SELECT 10 0 0
Empty boundary (start + finish) 30 192 1
Boundary, 1 tx, 2 writes (healthy path) 43 192 1
Boundary, 8 distinct write txs (overflow) 165 384 3

These numbers are not just documented but enforced: bench_test.go includes testing.AllocsPerRun guards (run by the normal go test) that fail if classification stops being allocation-free or a boundary starts allocating more than once, so a regression breaks CI rather than silently costing GC work in production.

Semantics and limitations

  • Rolled-back transactions still count as a unit. If a boundary runs tx A (rolled back) and then an auto-commit write, the boundary is structurally non-atomic and is reported.
  • Statement classification is heuristic (leading-keyword based, with token scanning for data-modifying CTEs). CALL/DO are conservatively treated as writes. Override with WithClassifier if needed.
  • Isolation problems are out of scope. A read-modify-write race (SELECT, compute, UPDATE without a transaction) involves only one write and is not detected. txnproof detects atomicity violations, not isolation violations.
  • Detached contexts are invisible. A write whose context does not carry the boundary is not attributed to it (use unbounded-write detection to at least see them).
  • MySQL's implicit commits are invisible client-side. MySQL implicitly commits an open transaction when a DDL statement (CREATE/ALTER/DROP, ...) runs inside it. txnproof's client-side transaction tracking is database-agnostic and does not model this: a boundary running BEGIN → write → DDL → write → COMMIT looks like one atomic unit to txnproof but actually spans three server-side transactions — a MySQL-specific false negative. Real-database tests with the mycheck cross-check do catch it: its state machine models implicit commits.
  • Cross-database writes are reported, not solved. If one boundary writes to two databases, that is ≥2 units by definition — which is exactly the point: a single SQL transaction cannot make it atomic, so the report tells you an outbox/saga/compensation is needed, or an allowlist entry with a reason.
  • Nested boundaries shadow. Starting a boundary on a context that already has one attributes subsequent statements to the inner boundary only — a deliberate contract (see nested boundaries); writes split across an outer and an inner boundary are therefore judged separately, never combined. Use WithNestedBoundaryDetection to surface nesting occurrences.

Examples

Runnable, zero-infrastructure examples (each backed by NewNullDB, so go run . works with no database) live under examples/:

  • examples/nethttp — net/http middleware that opens a boundary per request, named after the method and http.ServeMux route pattern (e.g. POST /users). Also shows AllowNonAtomicHere marking an intentional exemption from inside a handler, where the boundary is started by the middleware.
  • examples/graphql — GraphQL resolver middleware (via graphql-go) that opens a boundary per resolver, named Mutation.createUser-style.

License

MIT

Author

moznion (moznion@mail.moznion.net)

Documentation

Overview

Package txnproof detects non-atomic SQL execution: multiple write statements that run inside one logical boundary (a use case, a request, a job) without being wrapped in a single database transaction.

It works as a database/sql driver middleware, so the same detector serves three modes: pure unit tests (via NewNullDB), tests against a real database, and continuous production monitoring (via pluggable Reporters).

Index

Constants

View Source 
const Version = "0.3.0"

Version is the released version of go-txnproof. It is kept in sync with the latest git tag by tagpr (https://github.com/Songmu/tagpr); do not edit it by hand.

Variables

This section is empty.

Functions

func AllowNonAtomicHere added in v0.2.0 

func AllowNonAtomicHere(ctx context.Context, reason string, exactWriteUnits ...int)

AllowNonAtomicHere marks the boundary in ctx as intentionally non-atomic, suppressing its Violation exactly like the AllowNonAtomic boundary option: the two differ only in where the exemption lives, never in what it can express. The reason and the optional exactWriteUnits mean the same thing there, down to falling through to the central Allowlist when the boundary finishes with a count the mark does not cover.

It exists because a boundary is usually started far away from the code that makes it non-atomic — in a middleware or a use-case entry point, while the reason for the extra write is at the extra write. Marking it there keeps the explanation next to the code it explains, and keeps that explanation running code rather than a comment: 

// The audit row is written outside the domain transaction on purpose, so a
// failing audit sink cannot roll back the business change (TICKET-123).
txnproof.AllowNonAtomicHere(ctx, "audit write is best-effort (TICKET-123)", 2)
_, err := db.ExecContext(ctx, "INSERT INTO audit ...")

The mark applies to the innermost boundary in ctx, and it does not matter when during the boundary's life it is called: the evaluation happens at Finish. The last mark wins, replacing any earlier one (including one made by the AllowNonAtomic option). Calling it with no boundary in ctx, or after the boundary has finished, does nothing — the same way statements executed outside any boundary are ignored. Missing boundary plumbing is caught by WithUnboundedWriteDetection, which reports the write this call precedes.

Rot prevention is unchanged: an allowed boundary that finishes with fewer than 2 write units notifies StaleAllowReporter. Marking at the write site tends to be the more durable of the two, since a mark on a conditional path exists only on the executions that reach it.

func SlogAttrs 

func SlogAttrs(attrs []BoundaryAttr) []slog.Attr

SlogAttrs converts boundary attrs to log/slog attrs, for reporters built on slog. SlogReporter already applies it to the attrs it receives.

Types

type Allowlist 

type Allowlist struct {
	// contains filtered or unexported fields
}

Allowlist suppresses violations for boundaries that are intentionally non-atomic (e.g. best-effort audit writes, writes spanning databases that a single transaction cannot cover).

To keep the list from rotting, every entry tracks whether it actually suppressed a violation; check UnusedEntries in CI and fail when an entry no longer matches anything (the same discipline as unused //nolint directives).

func NewAllowlist 

func NewAllowlist() *Allowlist

NewAllowlist creates an empty Allowlist.

func (*Allowlist) Add 

func (a *Allowlist) Add(boundary, reason string, exactWriteUnits ...int) *Allowlist

Add registers a boundary name as intentionally non-atomic. The reason should say why and reference a ticket. Returns the Allowlist for chaining.

The optional exactWriteUnits pin how much non-atomicity the entry covers, exactly as for the in-code AllowNonAtomic mark: the entry then suppresses only boundaries finishing with one of the given write-unit counts, and any other count is reported as a Violation. Pass several counts for a boundary whose write count legitimately differs per code path. A write unit is one transaction that contained at least one write, or one auto-commit write (the same number reported as Violation.WriteUnits), so counts below 2 can never match a violation and leave the entry permanently unused.

func (*Allowlist) UnusedEntries 

func (a *Allowlist) UnusedEntries() []string

UnusedEntries returns the boundary names that never suppressed a violation, sorted. A non-empty result in CI means the allowlist has stale entries that should be removed — or, for an entry constrained to exact write-unit counts, that the boundary now violates with a count the entry does not cover (it is then reported as a Violation as well, and the entry needs reviewing rather than deleting).

type Baseline 

type Baseline struct {
	// contains filtered or unexported fields
}

Baseline is the ratchet helper for adopting txnproof on an existing codebase: capture the current violations once (BaselineFromViolations + Save), commit the file, and from then on only new violations fail — baselined boundaries are tolerated until fixed.

Entries are keyed on the boundary name alone. Write-unit counts and statement text vary by data and code path, so they would make the baseline unstable across runs; the boundary name is the stable identifier.

To keep the ratchet going down, every entry tracks whether it actually suppressed a violation; check UnusedEntries in CI and fail when an entry no longer matches anything — the same discipline as Allowlist.UnusedEntries.

func BaselineFromViolations 

func BaselineFromViolations(vs []Violation) *Baseline

BaselineFromViolations builds a Baseline from the boundary names of the given violations (typically CollectingReporter.Violations after a full run without any baseline installed). Duplicate boundaries collapse into one entry.

func LoadBaseline 

func LoadBaseline(path string) (*Baseline, error)

LoadBaseline reads a baseline file written by Save. A missing file is an error (check with errors.Is against fs.ErrNotExist): creating the baseline must stay a deliberate Save call, not a silent fallback.

func NewBaseline 

func NewBaseline() *Baseline

NewBaseline creates an empty Baseline.

func (*Baseline) Add 

func (b *Baseline) Add(boundary string) *Baseline

Add registers a boundary name in the baseline. Returns the Baseline for chaining. Prefer BaselineFromViolations + Save for the normal adoption flow; Add exists for programmatic construction.

func (*Baseline) Boundaries 

func (b *Baseline) Boundaries() []string

Boundaries returns the baselined boundary names, sorted.

func (*Baseline) Save 

func (b *Baseline) Save(path string) error

Save writes the baseline to path as deterministic, human-readable JSON: indented, boundaries sorted, with a comment field explaining the file, so diffs stay clean. Call it deliberately — on first adoption and on intentional regeneration — never on every run.

func (*Baseline) UnusedEntries 

func (b *Baseline) UnusedEntries() []string

UnusedEntries returns the baselined boundary names that never suppressed a violation, sorted. A non-empty result in CI means those boundaries are fixed: remove their entries from the baseline file so the ratchet keeps going down.

type BaselineReporter 

type BaselineReporter struct {
	// contains filtered or unexported fields
}

BaselineReporter filters violations through a Baseline before forwarding them to the wrapped Reporter: violations of baselined boundaries are swallowed (marking the entry used), everything else passes through. Unbounded-write and stale-allow reports are never baselined and are forwarded unchanged when the wrapped Reporter implements the corresponding interfaces.

func NewBaselineReporter 

func NewBaselineReporter(baseline *Baseline, next Reporter) *BaselineReporter

NewBaselineReporter wraps next so that violations of boundaries in baseline are suppressed. A nil baseline suppresses nothing.

func (*BaselineReporter) Report 

func (r *BaselineReporter) Report(ctx context.Context, v Violation)

func (*BaselineReporter) ReportNestedBoundary 

func (r *BaselineReporter) ReportNestedBoundary(ctx context.Context, n NestedBoundary)

func (*BaselineReporter) ReportStaleAllow 

func (r *BaselineReporter) ReportStaleAllow(ctx context.Context, s StaleAllow)

func (*BaselineReporter) ReportUnboundedWrite 

func (r *BaselineReporter) ReportUnboundedWrite(ctx context.Context, s StatementRecord)

type Boundary added in v0.1.0 

type Boundary struct {
	// contains filtered or unexported fields
}

Boundary is a live logical boundary (a use case, a request, a job): it accumulates the statement timeline of one execution and is finished by calling Finish. StartBoundary returns it both as the context to propagate and as the handle to finish.

It implements context.Context itself so that StartBoundary can return the boundary directly as the context node instead of wrapping it in a separate context.WithValue allocation: the boundary doubles as its own value carrier, exactly as the standard library's *valueCtx stores its parent. parent is the context the boundary was started on.

func (*Boundary) Deadline added in v0.1.0 

func (b *Boundary) Deadline() (time.Time, bool)

func (*Boundary) Done added in v0.1.0 

func (b *Boundary) Done() <-chan struct{}

func (*Boundary) Err added in v0.1.0 

func (b *Boundary) Err() error

func (*Boundary) Finish added in v0.1.0 

func (b *Boundary) Finish()

Finish evaluates the boundary and reports a Violation if its writes span two or more atomic units. Call it exactly when the boundary ends (typically via defer); it is idempotent.

func (*Boundary) Value added in v0.1.0 

func (b *Boundary) Value(key any) any

type BoundaryAttr 

type BoundaryAttr struct {
	Key   string
	Value any
}

BoundaryAttr is one string-keyed contextual value attached to a boundary and carried into every Violation the boundary produces. Use it to tie a report back to the execution that produced it (trace ID, request ID, user ID).

func Attr 

func Attr(key string, value any) BoundaryAttr

Attr constructs a BoundaryAttr.

type BoundaryOption 

type BoundaryOption func(*Boundary)

BoundaryOption configures a single boundary at StartBoundary / InBoundary.

func AllowNonAtomic 

func AllowNonAtomic(reason string, exactWriteUnits ...int) BoundaryOption

AllowNonAtomic marks the boundary as intentionally non-atomic, suppressing its Violation at the call site — the in-code alternative to a central Allowlist entry. The reason should say why and reference a ticket.

The optional exactWriteUnits pin how much non-atomicity the mark covers: the allow then applies only when the boundary finishes with exactly one of the given write-unit counts, and any other count is reported as a Violation as if the boundary carried no mark at all (the central Allowlist is still consulted afterwards). It keeps a reviewed exemption from silently growing as the boundary accumulates writes: 

// exactly the domain write plus the audit write, nothing more
txnproof.AllowNonAtomic("audit writes are best-effort (TICKET-123)", 2)

Passing several counts allows each of them, for a boundary whose write count legitimately differs per code path. A write unit is one transaction that contained at least one write, or one auto-commit write — the same number reported as Violation.WriteUnits, not the number of transactions — so counts below 2 can never match a violation and make the mark permanently stale.

Rot prevention works per execution instead of per entry: when an allowed boundary finishes with fewer than 2 write units (i.e. the allow suppressed nothing), reporters that implement StaleAllowReporter are notified — the same discipline as unused //nolint directives. A count the mark does not cover needs no such signal: it surfaces as the Violation itself.

AllowNonAtomicHere marks the same thing from the site of the write instead of from the boundary start.

func WithBoundaryAttrs 

func WithBoundaryAttrs(attrs ...BoundaryAttr) BoundaryOption

WithBoundaryAttrs attaches static attrs to a single boundary at StartBoundary / InBoundary — for values the caller already has at hand: 

ctx, b := detector.StartBoundary(ctx, "CreateUser",
	txnproof.WithBoundaryAttrs(txnproof.Attr("user_id", userID)))

They are appended after any attrs produced by WithBoundaryAttrsFunc. Duplicate keys are kept in order, never deduplicated.

type Classifier 

type Classifier func(query string) StatementKind

Classifier decides the StatementKind of a raw SQL string.

type CollectingReporter 

type CollectingReporter struct {
	// contains filtered or unexported fields
}

CollectingReporter accumulates violations in memory. Intended for tests.

func NewCollectingReporter 

func NewCollectingReporter() *CollectingReporter

NewCollectingReporter creates an empty CollectingReporter.

func (*CollectingReporter) NestedBoundaries 

func (r *CollectingReporter) NestedBoundaries() []NestedBoundary

NestedBoundaries returns a copy of the collected nested-boundary occurrences.

func (*CollectingReporter) Report 

func (r *CollectingReporter) Report(_ context.Context, v Violation)

func (*CollectingReporter) ReportNestedBoundary 

func (r *CollectingReporter) ReportNestedBoundary(_ context.Context, n NestedBoundary)

func (*CollectingReporter) ReportStaleAllow 

func (r *CollectingReporter) ReportStaleAllow(_ context.Context, s StaleAllow)

func (*CollectingReporter) ReportUnboundedWrite 

func (r *CollectingReporter) ReportUnboundedWrite(_ context.Context, s StatementRecord)

func (*CollectingReporter) RequireNoNestedBoundaries 

func (r *CollectingReporter) RequireNoNestedBoundaries(t TestingT)

RequireNoNestedBoundaries fails the test with one error per collected nested-boundary occurrence, enforcing that instrumentation layers do not overlap (requires WithNestedBoundaryDetection).

func (*CollectingReporter) RequireNoStaleAllows 

func (r *CollectingReporter) RequireNoStaleAllows(t TestingT)

RequireNoStaleAllows fails the test with one error per stale AllowNonAtomic mark, keeping in-code allows subject to the same rot discipline as Allowlist.UnusedEntries.

func (*CollectingReporter) RequireNoUnboundedWrites 

func (r *CollectingReporter) RequireNoUnboundedWrites(t TestingT)

RequireNoUnboundedWrites fails the test with one error per collected unbounded write, enforcing that every write in the exercised code ran with a boundary in its context (requires WithUnboundedWriteDetection).

func (*CollectingReporter) RequireNoViolations 

func (r *CollectingReporter) RequireNoViolations(t TestingT)

RequireNoViolations fails the test with one error per collected violation.

func (*CollectingReporter) Reset 

func (r *CollectingReporter) Reset()

Reset clears everything collected so far.

func (*CollectingReporter) StaleAllows 

func (r *CollectingReporter) StaleAllows() []StaleAllow

StaleAllows returns a copy of the collected stale AllowNonAtomic reports.

func (*CollectingReporter) UnboundedWrites 

func (r *CollectingReporter) UnboundedWrites() []StatementRecord

UnboundedWrites returns a copy of the collected unbounded write statements.

func (*CollectingReporter) Violations 

func (r *CollectingReporter) Violations() []Violation

Violations returns a copy of the collected violations.

type Detector 

type Detector struct {
	// contains filtered or unexported fields
}

Detector is the core of txnproof. Wrap a driver (or connector) with it, mark logical boundaries with StartBoundary / InBoundary, and it reports a Violation whenever a boundary executes two or more write statements that do not share a single transaction.

func New 

func New(opts ...Option) *Detector

New creates a Detector.

func (*Detector) InBoundary 

func (d *Detector) InBoundary(ctx context.Context, name string, f func(context.Context) error, opts ...BoundaryOption) error

InBoundary runs f inside a boundary and finishes it when f returns.

func (*Detector) NewNullDB 

func (d *Detector) NewNullDB() *sql.DB

NewNullDB returns a *sql.DB backed by an in-memory no-op driver wrapped by the Detector. Every statement succeeds and returns no rows; only the statement/transaction timeline is observed.

This is the sqlmock-free way to unit-test atomicity: inject the returned DB where your code expects a *sql.DB, run the use case inside a boundary, and assert no violations were reported. Unlike sqlmock, no expectations need to be declared.

func (*Detector) NewSession added in v0.3.0 

func (d *Detector) NewSession() *Session

NewSession creates a Session bound to the Detector. One Session per connection; see the type comment for the contract.

func (*Detector) StartBoundary 

func (d *Detector) StartBoundary(ctx context.Context, name string, opts ...BoundaryOption) (context.Context, *Boundary)

StartBoundary marks the beginning of a logical boundary (a use case, a request handler, a job) on the context. Every statement executed through a wrapped driver with the returned context is attributed to this boundary.

It returns the boundary both as the context to propagate and as the *Boundary handle to finish. Call Boundary.Finish exactly when the boundary ends (typically via defer) to evaluate it and report a Violation if its writes span two or more atomic units; Finish is idempotent.

Starting a boundary on a context that already carries one shadows the outer boundary for statements executed with the new context (reported when WithNestedBoundaryDetection is on).

func (*Detector) Wrap 

func (d *Detector) Wrap(drv driver.Driver) driver.Driver

Wrap wraps a database/sql driver so that every statement executed through it is observed by the Detector. Register the result with sql.Register: 

sql.Register("pgx-txnproof", detector.Wrap(stdlib.GetDefaultDriver()))
db, err := sql.Open("pgx-txnproof", dsn)

func (*Detector) WrapConnector 

func (d *Detector) WrapConnector(c driver.Connector) driver.Connector

WrapConnector wraps a driver.Connector for use with sql.OpenDB.

type NestedBoundary 

type NestedBoundary struct {
	// Outer is the name of the boundary that was already on the context.
	Outer string
	// Inner is the name of the newly started, shadowing boundary.
	Inner string
	// Time is when the inner boundary was started.
	Time time.Time
}

NestedBoundary is reported when a boundary is started on a context that already carries one (requires WithNestedBoundaryDetection). The shadow semantics are unchanged — statements attribute to the inner boundary only — so a nesting occurrence is not a Violation but a coverage signal: it usually means two instrumentation layers overlap (e.g. a resolver middleware and a use-case middleware both start boundaries).

type NestedBoundaryReporter 

type NestedBoundaryReporter interface {
	ReportNestedBoundary(ctx context.Context, n NestedBoundary)
}

NestedBoundaryReporter is an optional extension a Reporter can implement to also receive nested-boundary occurrences (requires WithNestedBoundaryDetection).

type Option 

type Option func(*Detector)

Option configures a Detector.

func WithAllowlist 

func WithAllowlist(a *Allowlist) Option

WithAllowlist installs an allowlist of boundary names whose violations are intentionally suppressed.

func WithBoundaryAttrsFunc 

func WithBoundaryAttrsFunc(f func(ctx context.Context) []BoundaryAttr) Option

WithBoundaryAttrsFunc installs a detector-level extractor that derives attrs from the context — the middleware-friendly way to stamp every boundary with trace/request IDs: set it up once and every Violation carries them for free.

f is evaluated once per boundary at StartBoundary (never per statement), with the context StartBoundary received; its attrs come first, followed by any per-boundary WithBoundaryAttrs. When unbounded-write detection is on, f is also evaluated once per unbounded write at record time (with the statement's context) and the result is delivered on the StatementRecord.

func WithClassifier 

func WithClassifier(c Classifier) Option

WithClassifier replaces DefaultClassifier for statement classification. The classifier must be a pure function of the query text: for statements executed through a prepared statement it is evaluated once at Prepare and the result is reused for every execution.

func WithMaxRecordedStatements 

func WithMaxRecordedStatements(n int) Option

WithMaxRecordedStatements caps how many statements are kept per boundary for violation reports (write-unit counting itself is never truncated). The default is 200.

func WithNestedBoundaryDetection 

func WithNestedBoundaryDetection() Option

WithNestedBoundaryDetection makes the detector notify reporters that implement NestedBoundaryReporter whenever a boundary is started on a context that already carries one. The shadow semantics are unchanged — statements still attribute to the inner boundary only — this option merely makes the nesting itself observable, so accidental double instrumentation (e.g. middleware at two layers) does not go unnoticed.

func WithReporter 

func WithReporter(rs ...Reporter) Option

WithReporter appends reporters that receive detected violations.

func WithUnboundedWriteDetection 

func WithUnboundedWriteDetection() Option

WithUnboundedWriteDetection makes the detector notify reporters that implement UnboundedWriteReporter about write statements executed with no boundary in their context (e.g. writes from detached goroutines).

type Reporter 

type Reporter interface {
	Report(ctx context.Context, v Violation)
}

Reporter receives detected violations. Implementations decide what to do: fail a test, log, emit a metric, notify an error tracker.

type ReporterFunc 

type ReporterFunc func(ctx context.Context, v Violation)

ReporterFunc adapts a function to the Reporter interface.

func (ReporterFunc) Report 

func (f ReporterFunc) Report(ctx context.Context, v Violation)

type Session added in v0.3.0 

type Session struct {
	// contains filtered or unexported fields
}

Session is the observation surface for one database connection that does not go through the database/sql driver stack: a native driver (pgx, a ClickHouse native client, ...) or an ORM hook that exposes statement text. The driver middleware itself is built on it, so both paths share one transaction-attribution state machine and one set of counting rules.

Create exactly one Session per underlying connection and use it the way the connection itself must be used: serially. That mirrors the guarantee database/sql gives a driver.Conn; a Session has no locking of its own. Statements from different connections must go to different Sessions — transaction attribution is per connection, and mixing connections in one Session would merge unrelated transactions into one unit.

Observe every statement the connection executes, at most once, with the context that carried it (that context is what attributes the statement to a boundary). Transaction control that the driver executes as statement text ("BEGIN"/"COMMIT"/"ROLLBACK" — pgx's Begin()/Commit() do exactly that) is tracked from the text alone; BeginTx/EndTx exist for transaction transitions that never surface as text.

Like the driver middleware, a Session observes submitted statements whether or not they later succeed: a failed write still proves a partial- write path structurally exists in the boundary. The one exception the middleware makes — ErrBadConn, where database/sql re-runs the statement on a fresh connection — is the integration's to make: skip observing a statement only when something else will observe its retry.

func (*Session) BeginTx added in v0.3.0 

func (s *Session) BeginTx()

BeginTx marks a transaction start that does not surface as statement text. Two callers need it: driver-API-level transactions (database/sql's ConnBeginTx, mirrored by the driver middleware), and protocol-level implicit transactions — a pgx batch is pipelined up to a single Sync, so PostgreSQL runs it as one implicit transaction even though no BEGIN is ever written; the integration brackets the batch with BeginTx/EndTx to count it as the single unit it is. Statements observed before the matching EndTx are attributed to this transaction.

Unlike a textual "BEGIN" (which is a no-op inside a transaction, matching the server's behavior), BeginTx trusts the caller and starts a new unit unconditionally — do not call it when the connection may already be in a transaction (a batch sent inside an explicit transaction belongs to that transaction; bracketing it would split the outer unit in two).

func (*Session) EndTx added in v0.3.0 

func (s *Session) EndTx()

EndTx marks the end of a transaction started by BeginTx — commit and rollback alike, since a rolled-back transaction still counts as a unit.

func (*Session) Observe added in v0.3.0 

func (s *Session) Observe(ctx context.Context, query string)

Observe records one executed statement, updating textual transaction state ("BEGIN"/"COMMIT"/"ROLLBACK" executed as statements) exactly like the driver middleware does.

type SlogReporter 

type SlogReporter struct {
	Logger *slog.Logger
}

SlogReporter reports violations through a *slog.Logger. Intended for production monitoring.

func NewSlogReporter 

func NewSlogReporter(l *slog.Logger) *SlogReporter

NewSlogReporter creates a SlogReporter. A nil logger means slog.Default().

func (*SlogReporter) Report 

func (r *SlogReporter) Report(ctx context.Context, v Violation)

func (*SlogReporter) ReportNestedBoundary 

func (r *SlogReporter) ReportNestedBoundary(ctx context.Context, n NestedBoundary)

func (*SlogReporter) ReportStaleAllow 

func (r *SlogReporter) ReportStaleAllow(ctx context.Context, s StaleAllow)

func (*SlogReporter) ReportUnboundedWrite 

func (r *SlogReporter) ReportUnboundedWrite(ctx context.Context, s StatementRecord)

type StaleAllow 

type StaleAllow struct {
	// Boundary is the name given to StartBoundary.
	Boundary string
	// Reason is the reason given to AllowNonAtomic.
	Reason string
	// WriteUnits is the number of atomic units the boundary actually used
	// (0 or 1).
	WriteUnits int
}

StaleAllow is reported when a boundary marked with AllowNonAtomic finishes without a violation to suppress: the allow did nothing for this execution. Note that this is per execution — a boundary whose write count varies by code path can legitimately produce both violations-suppressed and StaleAllow reports.

type StaleAllowReporter 

type StaleAllowReporter interface {
	ReportStaleAllow(ctx context.Context, s StaleAllow)
}

StaleAllowReporter is an optional extension a Reporter can implement to receive stale AllowNonAtomic marks (see StaleAllow).

type StatementKind 

type StatementKind int

StatementKind is the coarse classification of a SQL statement that txnproof cares about for atomicity tracking. 

const (
	// KindOther is a statement that is neither a read, a write, nor a
	// transaction-control statement (e.g. SET, SAVEPOINT, LOCK).
	KindOther StatementKind = iota
	// KindRead is a statement that does not modify data.
	KindRead
	// KindWrite is a statement that modifies data.
	KindWrite
	// KindBegin starts a transaction (textual BEGIN / START TRANSACTION).
	KindBegin
	// KindCommit commits a transaction (textual COMMIT / END).
	KindCommit
	// KindRollback rolls back a transaction (textual ROLLBACK / ABORT).
	KindRollback
)

func DefaultClassifier 

func DefaultClassifier(query string) StatementKind

DefaultClassifier classifies a statement by its leading keyword, skipping leading whitespace and SQL comments. It is a heuristic:

  • DML (INSERT/UPDATE/DELETE/MERGE/...), DDL (CREATE/ALTER/DROP/...), and procedure calls (CALL/DO) are treated as writes. Procedure calls are classified conservatively because their body is opaque.
  • WITH-prefixed statements are scanned for embedded write keywords so that data-modifying CTEs (WITH ... INSERT/UPDATE/DELETE) count as writes. The scan is token-based and may misfire on write keywords inside string literals; override with WithClassifier if this matters for your queries.
  • EXPLAIN is treated as a read even though EXPLAIN ANALYZE executes the inner statement.

func (StatementKind) String 

func (k StatementKind) String() string

type StatementRecord 

type StatementRecord struct {
	Query string
	Kind  StatementKind
	// TxID identifies the driver-level transaction the statement ran in.
	// 0 means the statement ran in auto-commit mode. IDs are process-local
	// sequence numbers, not database transaction IDs.
	TxID uint64
	Time time.Time
	// Attrs is populated only on records delivered to
	// UnboundedWriteReporter, with the result of WithBoundaryAttrsFunc
	// evaluated against the statement's context at record time. Records in
	// Violation.Statements leave it nil — the boundary's attrs live on the
	// Violation itself.
	Attrs []BoundaryAttr
}

StatementRecord is one SQL statement observed inside a boundary.

type TestingT 

type TestingT interface {
	Helper()
	Errorf(format string, args ...any)
}

TestingT is the subset of *testing.T that txnproof's test helpers need.

type ThrottlingReporter 

type ThrottlingReporter struct {
	// contains filtered or unexported fields
}

ThrottlingReporter wraps another Reporter and deduplicates repeated reports, so that a violating boundary on a hot path does not fire the wrapped reporter on every request. Intended for production monitoring.

Per boundary name, the first Violation is forwarded to the wrapped reporter immediately; subsequent Violations for the same boundary within the configured interval are suppressed; once the interval has elapsed, the next Violation is forwarded again and a new interval starts.

The optional reporter extensions are throttled with the same interval but with their own keys and independent windows:

  • Unbounded writes (UnboundedWriteReporter) are throttled per statement, keyed by the whitespace-normalized query text (truncated, see unboundedWriteKeyLen) — a hot-path unbounded write repeats the same statement text, so the statement is the natural dedup unit.
  • Stale AllowNonAtomic marks (StaleAllowReporter) are throttled per boundary name, independently of that boundary's Violation window — stale-allow reports are per execution and therefore just as noisy on a hot path as violations.
  • Nested boundaries (NestedBoundaryReporter) are throttled per outer/inner name pair — overlapping instrumentation layers repeat the same pair on every request.

Each extension is forwarded only when the wrapped reporter implements the corresponding interface, so wrapping neither swallows nor fabricates those signals.

Suppressed reports are not silently lost: cumulative per-key suppression counts are available via SuppressedViolations, SuppressedUnboundedWrites, and SuppressedStaleAllows, meant to be polled periodically (e.g. logged or exported as metrics on a ticker) to recover the true report volume.

Memory stays bounded: the two boundary-keyed maps grow with the set of boundary names, which is code-defined and small in practice; the statement-keyed map is capped at maxUnboundedWriteKeys.

func NewThrottlingReporter 

func NewThrottlingReporter(next Reporter, interval time.Duration) *ThrottlingReporter

NewThrottlingReporter wraps next so that repeated reports for the same key (boundary name for violations and stale allows, statement text for unbounded writes) are forwarded at most once per interval. A non-positive interval disables throttling: every report is forwarded.

func (*ThrottlingReporter) Report 

func (r *ThrottlingReporter) Report(ctx context.Context, v Violation)

Report forwards the first Violation per boundary immediately and at most one more per interval afterwards; the rest are counted as suppressed.

func (*ThrottlingReporter) ReportNestedBoundary 

func (r *ThrottlingReporter) ReportNestedBoundary(ctx context.Context, n NestedBoundary)

ReportNestedBoundary forwards nested-boundary occurrences throttled per outer/inner name pair. It is a no-op when the wrapped reporter does not implement NestedBoundaryReporter.

func (*ThrottlingReporter) ReportStaleAllow 

func (r *ThrottlingReporter) ReportStaleAllow(ctx context.Context, s StaleAllow)

ReportStaleAllow forwards stale AllowNonAtomic reports throttled per boundary name (independently of the boundary's Violation window). It is a no-op when the wrapped reporter does not implement StaleAllowReporter.

func (*ThrottlingReporter) ReportUnboundedWrite 

func (r *ThrottlingReporter) ReportUnboundedWrite(ctx context.Context, s StatementRecord)

ReportUnboundedWrite forwards unbounded write reports throttled per statement text. It is a no-op when the wrapped reporter does not implement UnboundedWriteReporter.

func (*ThrottlingReporter) SuppressedNestedBoundaries 

func (r *ThrottlingReporter) SuppressedNestedBoundaries() map[string]int

SuppressedNestedBoundaries returns the cumulative number of suppressed nested-boundary reports per "outer\x00inner" name pair.

func (*ThrottlingReporter) SuppressedStaleAllows 

func (r *ThrottlingReporter) SuppressedStaleAllows() map[string]int

SuppressedStaleAllows returns the cumulative number of suppressed stale AllowNonAtomic reports per boundary name since the reporter was created.

func (*ThrottlingReporter) SuppressedUnboundedWrites 

func (r *ThrottlingReporter) SuppressedUnboundedWrites() map[string]int

SuppressedUnboundedWrites returns the cumulative number of suppressed unbounded write reports per statement key (whitespace-normalized, possibly truncated query text) since the reporter was created.

func (*ThrottlingReporter) SuppressedViolations 

func (r *ThrottlingReporter) SuppressedViolations() map[string]int

SuppressedViolations returns the cumulative number of suppressed Violations per boundary name since the reporter was created. Counts only grow; boundaries with zero suppressions are omitted. Poll it periodically to recover the true violation volume behind the throttled stream.

type UnboundedWriteReporter 

type UnboundedWriteReporter interface {
	ReportUnboundedWrite(ctx context.Context, s StatementRecord)
}

UnboundedWriteReporter is an optional extension a Reporter can implement to also receive write statements executed with no boundary in their context (requires WithUnboundedWriteDetection).

type Violation 

type Violation struct {
	// Boundary is the name given to StartBoundary.
	Boundary string
	// WriteUnits is the number of distinct atomic units that contained
	// writes. Atomic execution means WriteUnits == 1.
	WriteUnits int
	// Statements is the recorded statement timeline of the boundary
	// (reads included), capped by WithMaxRecordedStatements.
	Statements []StatementRecord
	// TruncatedStatements is how many statements were dropped from
	// Statements due to the cap.
	TruncatedStatements int
	// AllowedWriteUnits are the exact write-unit counts an AllowNonAtomic
	// mark (or Allowlist entry) covered for this boundary, set only when the
	// boundary was marked but finished with a count outside them — i.e. this
	// violation is reported *because* the reviewed count no longer matches.
	// It is nil for an ordinary, unmarked violation.
	AllowedWriteUnits []int
	// Attrs is the contextual metadata attached to the boundary (trace ID,
	// request ID, ...): the result of WithBoundaryAttrsFunc evaluated at
	// boundary start, followed by any WithBoundaryAttrs entries. Duplicate
	// keys are kept in order.
	Attrs []BoundaryAttr
}

Violation is reported when a boundary's write statements span two or more atomic units (distinct transactions and/or auto-commit statements), meaning the boundary is not atomic: a crash between units leaves partial state.

func (Violation) String 

func (v Violation) String() string

String renders a human-readable multi-line summary listing the write statements grouped by atomic unit.

Source Files

View all Source files

Directories

Path Synopsis
Package crosscheck verifies a scenario's atomicity from a database server's own record of execution: given the statements the server logged while a test scenario ran, each annotated with the server-side transaction it ran in, it checks that all write statements shared one transaction.
 Package crosscheck verifies a scenario's atomicity from a database server's own record of execution: given the statements the server logged while a test scenario ran, each annotated with the server-side transaction it ran in, it checks that all write statements shared one transaction.
Package mycheck is the MySQL adapter for the crosscheck package: it parses the general query log a MySQL server produced while a test scenario ran, reconstructs which transaction each logged statement ran in, and delegates to crosscheck to verify that all write statements shared one transaction.
 Package mycheck is the MySQL adapter for the crosscheck package: it parses the general query log a MySQL server produced while a test scenario ran, reconstructs which transaction each logged statement ran in, and delegates to crosscheck to verify that all write statements shared one transaction.
Package pgcheck is the PostgreSQL adapter for the crosscheck package: it parses the log lines a PostgreSQL server produced while a test scenario ran, maps each logged statement to the server-side transaction it ran in, and delegates to crosscheck to verify that all write statements shared one transaction.
 Package pgcheck is the PostgreSQL adapter for the crosscheck package: it parses the log lines a PostgreSQL server produced while a test scenario ran, maps each logged statement to the server-side transaction it ran in, and delegates to crosscheck to verify that all write statements shared one transaction.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.