diff --git a/support/connection/registry.go b/support/connection/registry.go index 5ba7b9b..ab88b21 100644 --- a/support/connection/registry.go +++ b/support/connection/registry.go @@ -94,3 +94,34 @@ func Managers() map[string]Manager { return ret } + +// GetId returns the shared-connection id `manager` was registered under, or "" when the +// manager is not a shared connection (for example an inline connection config). It mirrors +// IsShared, which already performs the same linear scan; `managers` holds one entry per app +// connection, so the scan is single-digit. +// +// A map[Manager]string reverse index was deliberately rejected: indexing by a manager panics +// with "hash of unhashable type" for a value-receiver manager over a struct holding a map or +// slice, which would be an unrecoverable panic at app startup for every app in the org, +// including apps that never use a transaction. The recover below closes the same pre-existing +// hazard IsShared already carries. +func GetId(manager Manager) (id string) { + if manager == nil { + return "" + } + + defer func() { + if r := recover(); r != nil { + log.RootLogger().Debugf("connection.GetId: manager is not comparable: %v", r) + id = "" + } + }() + + for cid, mgr := range managers { + if manager == mgr { + return cid + } + } + + return "" +} diff --git a/support/sqltx/fakedriver_test.go b/support/sqltx/fakedriver_test.go new file mode 100644 index 0000000..8aec2f8 --- /dev/null +++ b/support/sqltx/fakedriver_test.go @@ -0,0 +1,162 @@ +package sqltx + +import ( + "context" + "database/sql" + "database/sql/driver" + "io" + "sync" + "sync/atomic" +) + +// A minimal in-memory database/sql driver. No real database is available, and adding a mock +// dependency to core is not acceptable, so the tests drive the genuine database/sql machinery - +// pooling, connection pinning, Tx.PrepareContext, Commit/Rollback - against this. +// +// It deliberately implements driver.ConnBeginTx and driver.ConnPrepareContext, because those are +// exactly the interfaces the real connectors' drivers implement (verified for mssql, mysql, +// godror and lib/pq) and they are what makes prepare-on-transaction work. + +type fakeDriver struct { + mu sync.Mutex + opened int32 // connections handed out, cumulative + live int32 // connections currently open + prepares int32 // PrepareContext calls, cumulative + commits int32 + rollbacks int32 +} + +func (d *fakeDriver) Open(string) (driver.Conn, error) { + atomic.AddInt32(&d.opened, 1) + atomic.AddInt32(&d.live, 1) + return &fakeConn{drv: d}, nil +} + +func (d *fakeDriver) counts() (opened, live, prepares, commits, rollbacks int32) { + return atomic.LoadInt32(&d.opened), atomic.LoadInt32(&d.live), + atomic.LoadInt32(&d.prepares), atomic.LoadInt32(&d.commits), + atomic.LoadInt32(&d.rollbacks) +} + +type fakeConn struct { + drv *fakeDriver + inTx bool + badTx bool +} + +var ( + _ driver.Conn = (*fakeConn)(nil) + _ driver.ConnBeginTx = (*fakeConn)(nil) + _ driver.ConnPrepareContext = (*fakeConn)(nil) +) + +func (c *fakeConn) Prepare(query string) (driver.Stmt, error) { + return c.PrepareContext(context.Background(), query) +} + +func (c *fakeConn) PrepareContext(_ context.Context, query string) (driver.Stmt, error) { + atomic.AddInt32(&c.drv.prepares, 1) + return &fakeStmt{conn: c, query: query}, nil +} + +func (c *fakeConn) Close() error { + atomic.AddInt32(&c.drv.live, -1) + return nil +} + +func (c *fakeConn) Begin() (driver.Tx, error) { + return c.BeginTx(context.Background(), driver.TxOptions{}) +} + +func (c *fakeConn) BeginTx(_ context.Context, _ driver.TxOptions) (driver.Tx, error) { + if c.inTx { + // Mirrors godror conn.go:337-341, which rejects a nested begin on the same connection. + return nil, driver.ErrBadConn + } + c.inTx = true + return &fakeTx{conn: c}, nil +} + +type fakeTx struct{ conn *fakeConn } + +func (t *fakeTx) Commit() error { + t.conn.inTx = false + atomic.AddInt32(&t.conn.drv.commits, 1) + if t.conn.badTx { + return driver.ErrBadConn + } + return nil +} + +func (t *fakeTx) Rollback() error { + t.conn.inTx = false + atomic.AddInt32(&t.conn.drv.rollbacks, 1) + return nil +} + +type fakeStmt struct { + conn *fakeConn + query string +} + +func (s *fakeStmt) Close() error { return nil } +func (s *fakeStmt) NumInput() int { return -1 } + +func (s *fakeStmt) Exec([]driver.Value) (driver.Result, error) { + return driver.RowsAffected(1), nil +} + +func (s *fakeStmt) Query([]driver.Value) (driver.Rows, error) { + return &fakeRows{cols: []string{"c"}, rows: [][]driver.Value{{int64(1)}}}, nil +} + +type fakeRows struct { + cols []string + rows [][]driver.Value + i int +} + +func (r *fakeRows) Columns() []string { return r.cols } +func (r *fakeRows) Close() error { return nil } + +func (r *fakeRows) Next(dest []driver.Value) error { + if r.i >= len(r.rows) { + return io.EOF + } + copy(dest, r.rows[r.i]) + r.i++ + return nil +} + +// newFakeDB registers a uniquely-named driver and opens a pool on it. The unique name matters: +// sql.Register panics on a duplicate, and these tests each want isolated counters. +var fakeSeq int32 + +func newFakeDB(maxOpen int) (*sql.DB, *fakeDriver) { + d := &fakeDriver{} + name := "sqltx-fake-" + itoa(int(atomic.AddInt32(&fakeSeq, 1))) + sql.Register(name, d) + + db, err := sql.Open(name, "") + if err != nil { + panic(err) + } + if maxOpen > 0 { + db.SetMaxOpenConns(maxOpen) + } + return db, d +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + var b [20]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} diff --git a/support/sqltx/handle.go b/support/sqltx/handle.go new file mode 100644 index 0000000..df27c06 --- /dev/null +++ b/support/sqltx/handle.go @@ -0,0 +1,374 @@ +package sqltx + +import ( + "context" + "database/sql" + "errors" + "fmt" + "sync" + "time" +) + +// ErrTxFinished is returned by PrepareCached after the finalizer has committed or rolled back. +// It WRAPS sql.ErrTxDone so connectors may use errors.Is(err, sql.ErrTxDone) while the +// user-facing text names the subflow instead of the opaque "sql: transaction has already been +// committed or rolled back". +var ErrTxFinished = fmt.Errorf( + "SUBFLOW-TX-005: the enclosing transactional subflow has already ended; this statement was not executed: %w", + sql.ErrTxDone) + +// maxCachedStmts bounds the per-transaction statement memo. +// +// The connectors' EvaluateQuery expands an IN-clause into a placeholder list sized to the input +// array, so a loop iterating with differently-sized arrays produces a DISTINCT SQL text per +// iteration. +// +// Be clear about what this fixes. It bounds the Go-side map only. Statements prepared on a +// *sql.Tx are appended to tx.stmts and are closed by Commit/Rollback, so N distinct SQL texts +// inside one transaction pin N server-side cursors REGARDLESS of memoisation - the ORA-01000 +// class of exhaustion is a property of the transaction's length, not of this cache. The cap +// stops the map growing without bound and records the fact so it is diagnosable; dynamic SQL in +// a loop inside a transactional subflow remains a documented anti-pattern. +const maxCachedStmts = 128 + +// Handle is the ambient transaction for exactly one transactional subflow invocation. +// +// THREE MUTEXES, ONE OF WHICH - mu - IS THE ONLY ONE A CONNECTOR EVER TOUCHES. A single mutex +// self-deadlocks: the connector takes the operation lock at the top of PreparedQuery and then +// calls getStatement, which re-enters the handle, and sync.Mutex is not reentrant. +// +// mu the D3 SERIALISATION lock. A connector takes it at the TOP of a whole Prepared* +// operation and releases it only after *sql.Rows has been fully drained, because +// database/sql pins the transaction's single connection until the Rows is closed. +// Exposed as Lock/Unlock/TryLockFor. +// stmtsMu guards the statement memo, the done flag and the warn set ONLY. +// opMu guards the in-flight statement-cancel registrations ONLY. +// +// The lock order is acyclic because nothing inside this package ever takes mu, and opMu and +// stmtsMu are never held simultaneously. That is what makes PrepareCached, MarkDone, IsDone, +// WarnOnce and OpContext safe to call with mu held, which is the normal case. +// +// Every method is nil-receiver safe: connectors call them on values that are nil on the +// non-transactional path. +type Handle struct { + connID string + db *sql.DB + tx *sql.Tx + + // baseCtx is the context the transaction was BEGUN on. Cancelling it makes database/sql roll + // the transaction back asynchronously, via the watchdog goroutine it starts in BeginTx. The + // ENGINE therefore never cancels it: the subflow activity roots it at context.Background(), + // and only the finalizer cancels it, AFTER Commit or Rollback. + baseCtx context.Context + + mu sync.Mutex // D3 operation lock + + stmtsMu sync.Mutex + stmts map[string]*sql.Stmt // memoised prepares, keyed by SQL TEXT + done bool + memoFull bool + warned map[string]bool // D13 dedup + memoFullWarned bool + + opMu sync.Mutex + opNextTok uint64 + opCancels map[uint64]context.CancelFunc +} + +// NewHandle builds a handle for a transaction that has already been begun. +func NewHandle(connID string, db *sql.DB, tx *sql.Tx, baseCtx context.Context) *Handle { + if baseCtx == nil { + baseCtx = context.Background() + } + + return &Handle{ + connID: connID, + db: db, + tx: tx, + baseCtx: baseCtx, + stmts: make(map[string]*sql.Stmt), + warned: make(map[string]bool), + } +} + +// ConnID is the connection id this transaction was opened on. +func (h *Handle) ConnID() string { + if h == nil { + return "" + } + return h.connID +} + +// DB is the pool the transaction was taken from. +func (h *Handle) DB() *sql.DB { + if h == nil { + return nil + } + return h.db +} + +// Tx is the underlying transaction. +func (h *Handle) Tx() *sql.Tx { + if h == nil { + return nil + } + return h.tx +} + +// Context returns the transaction's own context, for read-only use such as a parent. To EXECUTE +// a statement use OpContext instead, so the finalizer can interrupt it. +func (h *Handle) Context() context.Context { + if h == nil { + return context.Background() + } + return h.baseCtx +} + +// Lock takes the D3 operation lock. It must wrap a WHOLE Prepared* operation, including full row +// drainage. +func (h *Handle) Lock() { + if h != nil { + h.mu.Lock() + } +} + +// Unlock releases the D3 operation lock. +func (h *Handle) Unlock() { + if h != nil { + h.mu.Unlock() + } +} + +// TryLockFor polls Mutex.TryLock until d elapses. It exists so the engine finalizer never blocks +// forever behind an abandoned execTimeout Eval goroutine. Acquiring the lock is BEST EFFORT on +// the rollback path: the finalizer rolls back either way, after CancelInFlight. +func (h *Handle) TryLockFor(d time.Duration) bool { + if h == nil { + return false + } + if h.mu.TryLock() { + return true + } + + deadline := time.Now().Add(d) + for { + time.Sleep(2 * time.Millisecond) + if h.mu.TryLock() { + return true + } + if !time.Now().Before(deadline) { + return false + } + } +} + +// NoRelease is the release func for a statement the caller does not own: the process-wide cache +// owns it on the non-enlisted path, and the transaction owns it on the memoised path. Callers +// always `defer release()`, so having a shared no-op keeps every call site the same shape. +var NoRelease = func() {} + +// PrepareCached prepares sqlText ON THE TRANSACTION and memoises it for the transaction's +// lifetime. +// +// On the enlisted path we must NEVER touch the pool. sql.DB.Prepare acquires a SECOND pooled +// connection while the transaction pins one; at maxOpenConnection:1 - an ordinary user setting - +// that is a hard deadlock with no deadline to break it. Tx.PrepareContext uses the transaction's +// already-pinned connection and appends to tx.stmts so Commit/Rollback closes it. +// +// The returned release func MUST be deferred by the caller. It is a no-op for a memoised +// statement, which the transaction owns and closes; it closes the statement only when the memo +// was at its cap and this statement was prepared for this call alone. Never Close the returned +// statement directly - use release. +// +// Safe to call with or without the operation lock held. +func (h *Handle) PrepareCached(sqlText string) (*sql.Stmt, func(), error) { + if h == nil { + return nil, NoRelease, errors.New("no ambient transaction") + } + // Pass the transaction's own context explicitly rather than nil. PrepareCachedContext still + // tolerates a nil context for callers that have none, but routing through it here would be a + // staticcheck SA1012 violation for no benefit. + return h.PrepareCachedContext(h.baseCtx, sqlText) +} + +// PrepareCachedContext is PrepareCached with an explicit context for the PREPARE itself. +func (h *Handle) PrepareCachedContext(ctx context.Context, sqlText string) (*sql.Stmt, func(), error) { + if h == nil || h.tx == nil { + return nil, NoRelease, errors.New("no ambient transaction") + } + if ctx == nil { + ctx = h.baseCtx + } + + h.stmtsMu.Lock() + if h.done { + h.stmtsMu.Unlock() + return nil, NoRelease, ErrTxFinished + } + if st, ok := h.stmts[sqlText]; ok { + h.stmtsMu.Unlock() + return st, NoRelease, nil + } + h.stmtsMu.Unlock() + + // PREPARE with no lock of ours held: mu may be held by our own caller, and MarkDone must not + // block behind network I/O. + st, err := h.tx.PrepareContext(ctx, sqlText) // on the TX - never on the pool + if err != nil { + return nil, NoRelease, err + } + + h.stmtsMu.Lock() + if h.done { // finalised while we were preparing + h.stmtsMu.Unlock() + _ = st.Close() + return nil, NoRelease, ErrTxFinished + } + if existing, ok := h.stmts[sqlText]; ok { // lost a benign race + h.stmtsMu.Unlock() + _ = st.Close() + return existing, NoRelease, nil + } + if len(h.stmts) >= maxCachedStmts { + // At the cap. Hand the statement over to the caller instead of memoising it, so a loop + // generating distinct SQL - an IN-clause expanded per batch size, say - closes each one + // as it goes rather than pinning a server-side cursor per iteration until the + // transaction ends. That exhaustion (ORA-01000 and friends) is a property of the + // transaction's length, which memoisation alone cannot bound. + h.memoFull = true + h.stmtsMu.Unlock() + return st, func() { _ = st.Close() }, nil + } + h.stmts[sqlText] = st + h.stmtsMu.Unlock() + + return st, NoRelease, nil +} + +// MemoSaturated reports whether the statement memo hit its cap, so the connector can emit one +// diagnostic. See maxCachedStmts for what this does and does not tell you. +func (h *Handle) MemoSaturated() bool { + if h == nil { + return false + } + h.stmtsMu.Lock() + defer h.stmtsMu.Unlock() + return h.memoFull +} + +// MemoSaturatedOnce is MemoSaturated, but true at most once per handle, for a one-shot log. +func (h *Handle) MemoSaturatedOnce() bool { + if h == nil { + return false + } + h.stmtsMu.Lock() + defer h.stmtsMu.Unlock() + if !h.memoFull || h.memoFullWarned { + return false + } + h.memoFullWarned = true + return true +} + +// OpContext returns the context a connector must use to EXECUTE a statement inside the +// transaction, plus its release func. +// +// The parent is baseCtx, NOT context.Background(): a per-statement +// context.WithTimeout(context.Background(), queryTimeout) firing mid-statement cancels the query +// on the connection the transaction is pinned to and typically poisons it. Cancelling baseCtx +// itself is not an option either - database/sql's watchdog would roll the transaction back. +// +// Registrations are TOKEN-KEYED. A single cancel slot loses B's registration when two operations +// overlap, and then A's release clears B's, leaving CancelInFlight unable to unwedge anything. +// Overlap should not happen under the D3 discipline, but the handle must not depend on +// discipline it cannot enforce. +// +// A timeout of zero or less means no deadline. +func (h *Handle) OpContext(timeout time.Duration) (context.Context, context.CancelFunc) { + if h == nil { + return context.Background(), func() {} + } + + var ( + ctx context.Context + cancel context.CancelFunc + ) + if timeout > 0 { + ctx, cancel = context.WithTimeout(h.baseCtx, timeout) + } else { + ctx, cancel = context.WithCancel(h.baseCtx) + } + + h.opMu.Lock() + h.opNextTok++ + tok := h.opNextTok + if h.opCancels == nil { + h.opCancels = make(map[uint64]context.CancelFunc, 2) + } + h.opCancels[tok] = cancel + h.opMu.Unlock() + + return ctx, func() { + cancel() + h.opMu.Lock() + delete(h.opCancels, tok) + h.opMu.Unlock() + } +} + +// CancelInFlight cancels EVERY statement context OpContext has handed out and not yet released. +// The finalizer calls it on the ROLLBACK path only. +func (h *Handle) CancelInFlight() { + if h == nil { + return + } + + h.opMu.Lock() + pending := make([]context.CancelFunc, 0, len(h.opCancels)) + for tok, c := range h.opCancels { + pending = append(pending, c) + delete(h.opCancels, tok) + } + h.opMu.Unlock() + + for _, c := range pending { + c() + } +} + +// MarkDone is called by the finalizer BEFORE Commit or Rollback. Afterwards PrepareCached returns +// ErrTxFinished rather than a statement the commit is about to close. +func (h *Handle) MarkDone() { + if h == nil { + return + } + h.stmtsMu.Lock() + h.done = true + h.stmtsMu.Unlock() +} + +// IsDone reports whether the transaction has been finalised. A nil handle counts as done. +func (h *Handle) IsDone() bool { + if h == nil { + return true + } + h.stmtsMu.Lock() + defer h.stmtsMu.Unlock() + return h.done +} + +// WarnOnce reports whether the caller should emit a warning for key on this handle. True exactly +// once per handle per key - that is, once per transactional subflow invocation per foreign +// connection, never process-wide. +func (h *Handle) WarnOnce(key string) bool { + if h == nil { + return false + } + h.stmtsMu.Lock() + defer h.stmtsMu.Unlock() + if h.warned[key] { + return false + } + h.warned[key] = true + return true +} diff --git a/support/sqltx/handle_test.go b/support/sqltx/handle_test.go new file mode 100644 index 0000000..c51f6ae --- /dev/null +++ b/support/sqltx/handle_test.go @@ -0,0 +1,446 @@ +package sqltx + +import ( + "context" + "database/sql" + "errors" + "strconv" + "sync" + "testing" + "time" +) + +// beginFake opens a pool with the given MaxOpenConns, begins a transaction and returns a handle +// over it, plus the driver's counters and a cleanup func. +func beginFake(t *testing.T, maxOpen int) (*Handle, *fakeDriver, func()) { + t.Helper() + + db, drv := newFakeDB(maxOpen) + base := context.Background() + + tx, err := db.BeginTx(base, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + + h := NewHandle("conn-a", db, tx, base) + + return h, drv, func() { + _ = tx.Rollback() + _ = db.Close() + } +} + +// TestX2_LockThenPrepareCachedDoesNotDeadlock is the regression test for blocker X2. +// +// With a single mutex serving both the D3 operation lock and the statement memo, this sequence +// self-deadlocks on the first enlisted statement: the connector takes the operation lock at the +// top of PreparedQuery, then getStatement re-enters the handle, and sync.Mutex is not reentrant. +// The failure mode is a silent hang, so the whole suite runs with -timeout. +func TestX2_LockThenPrepareCachedDoesNotDeadlock(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + done := make(chan struct{}) + + go func() { + defer close(done) + h.Lock() + defer h.Unlock() + + if _, _, err := h.PrepareCached("SELECT 1"); err != nil { + t.Errorf("first PrepareCached with the operation lock held: %v", err) + } + if _, _, err := h.PrepareCached("SELECT 1"); err != nil { + t.Errorf("second PrepareCached (memo hit) with the operation lock held: %v", err) + } + // The other lock-held-safe methods, same reasoning. + h.IsDone() + h.WarnOnce("k") + h.MemoSaturated() + relCtx, release := h.OpContext(0) + _ = relCtx + release() + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("deadlocked: PrepareCached re-entered the operation lock") + } +} + +// TestX3_PrepareCachedNeverTouchesThePool is the regression test for blocker X3. +// +// A cache-miss db.Prepare needs a SECOND pooled connection while the transaction pins one. At +// maxOpenConnection:1 - an ordinary user setting on every connector's connection tile - that is +// a hard deadlock on context.Background() with no deadline to break it. Preparing on the +// transaction uses the already-pinned connection instead. +func TestX3_PrepareCachedNeverTouchesThePool(t *testing.T) { + h, drv, cleanup := beginFake(t, 1) // the pool has exactly one connection, and the tx holds it + defer cleanup() + + done := make(chan error, 1) + go func() { + _, _, err := h.PrepareCached("INSERT INTO t VALUES (?)") + done <- err + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("PrepareCached at MaxOpenConns=1: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("deadlocked at MaxOpenConns=1: the enlisted path acquired a pooled connection") + } + + if opened, _, _, _, _ := drv.counts(); opened != 1 { + t.Fatalf("driver opened %d connections, want exactly 1 (the transaction's)", opened) + } +} + +// TestX3_PoolPrepareWouldDeadlock documents the hazard X3 describes, so the reasoning behind +// PrepareCached is verifiable rather than asserted. It is the behaviour we must NOT have. +func TestX3_PoolPrepareWouldDeadlock(t *testing.T) { + db, _ := newFakeDB(1) + defer db.Close() + + tx, err := db.BeginTx(context.Background(), nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + defer tx.Rollback() + + blocked := make(chan struct{}) + go func() { + // The tx holds the only connection, so this waits for one that cannot be returned. + stmt, err := db.Prepare("SELECT 1") + if err == nil { + _ = stmt.Close() + } + close(blocked) + }() + + select { + case <-blocked: + t.Fatal("expected db.Prepare to block while the transaction pins the only connection; " + + "if this now returns, re-examine whether PrepareCached still needs to avoid the pool") + case <-time.After(300 * time.Millisecond): + // Blocked, as expected. + } +} + +func TestPrepareCachedMemoisesBySQLText(t *testing.T) { + h, drv, cleanup := beginFake(t, 0) + defer cleanup() + + first, _, err := h.PrepareCached("SELECT 1") + if err != nil { + t.Fatalf("first: %v", err) + } + second, _, err := h.PrepareCached("SELECT 1") + if err != nil { + t.Fatalf("second: %v", err) + } + + if first != second { + t.Fatal("PrepareCached must return the memoised statement for identical SQL text") + } + if _, _, prepares, _, _ := drv.counts(); prepares != 1 { + t.Fatalf("driver prepared %d times, want 1", prepares) + } +} + +// TestMemoDoesNotGrowWithoutBound covers the dynamic-SQL loop: a connector expanding an IN-clause +// produces a distinct SQL text per iteration. +func TestMemoDoesNotGrowWithoutBound(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + for i := 0; i < 200; i++ { + if _, _, err := h.PrepareCached("SELECT " + strconv.Itoa(i)); err != nil { + t.Fatalf("PrepareCached #%d: %v", i, err) + } + } + + h.stmtsMu.Lock() + n := len(h.stmts) + h.stmtsMu.Unlock() + + if n != maxCachedStmts { + t.Fatalf("memo holds %d statements, want the cap of %d", n, maxCachedStmts) + } + if !h.MemoSaturated() { + t.Fatal("MemoSaturated should report true once the cap is hit") + } + if !h.MemoSaturatedOnce() { + t.Fatal("MemoSaturatedOnce should report true the first time") + } + if h.MemoSaturatedOnce() { + t.Fatal("MemoSaturatedOnce should report false the second time") + } +} + +// TestReleaseIsNoOpWhenMemoisedButClosesOnOverflow pins the contract the connectors rely on. +// +// A memoised statement is owned by the transaction and must NOT be closed by the caller - +// Commit/Rollback closes it. A statement handed out after the memo hit its cap is owned by the +// CALLER, and its release must close it, so a loop generating distinct SQL does not pin a +// server-side cursor per iteration for the rest of the transaction. +func TestReleaseIsNoOpWhenMemoisedButClosesOnOverflow(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + // Memoised: release must be the shared no-op, and the statement must stay usable afterwards. + st, release, err := h.PrepareCached("SELECT 1") + if err != nil { + t.Fatalf("memoised prepare: %v", err) + } + release() + again, _, err := h.PrepareCached("SELECT 1") + if err != nil { + t.Fatalf("after release: %v", err) + } + if again != st { + t.Fatal("release() must not have discarded the memoised statement") + } + + // Fill the memo, then verify the overflow statement really is closed by its release. + for i := 0; i < maxCachedStmts+5; i++ { + _, rel, err := h.PrepareCached("SELECT over " + strconv.Itoa(i)) + if err != nil { + t.Fatalf("overflow prepare #%d: %v", i, err) + } + rel() + } + if !h.MemoSaturated() { + t.Fatal("the memo should be saturated by now") + } + + overflowStmt, rel, err := h.PrepareCached("SELECT the-overflow-one") + if err != nil { + t.Fatalf("overflow prepare: %v", err) + } + rel() + // Closing twice is the observable proof the release closed it: a second Close on an already + // closed *sql.Stmt is a no-op returning nil, whereas a live memoised statement would still be + // present in h.stmts. Assert on the memo instead, which is unambiguous. + h.stmtsMu.Lock() + _, memoised := h.stmts["SELECT the-overflow-one"] + n := len(h.stmts) + h.stmtsMu.Unlock() + if memoised { + t.Fatal("an overflow statement must not be memoised") + } + if n != maxCachedStmts { + t.Fatalf("memo holds %d, want it pinned at the cap %d", n, maxCachedStmts) + } + if overflowStmt == nil { + t.Fatal("the overflow statement should still have been usable before release") + } +} + +func TestPrepareCachedAfterMarkDoneReturnsErrTxFinished(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + h.MarkDone() + + _, _, err := h.PrepareCached("SELECT 1") + if !errors.Is(err, ErrTxFinished) { + t.Fatalf("err = %v, want ErrTxFinished", err) + } + // Connectors branch on the stdlib sentinel, so the wrap must survive. + if !errors.Is(err, sql.ErrTxDone) { + t.Fatalf("ErrTxFinished must wrap sql.ErrTxDone; got %v", err) + } + if !h.IsDone() { + t.Fatal("IsDone should be true after MarkDone") + } +} + +func TestNilHandleIsSafeEverywhere(t *testing.T) { + var h *Handle // the non-transactional path + + if h.ConnID() != "" || h.DB() != nil || h.Tx() != nil { + t.Fatal("nil handle accessors should return zero values") + } + if !h.IsDone() { + t.Fatal("a nil handle counts as done") + } + if h.WarnOnce("k") { + t.Fatal("a nil handle must not ask the caller to warn") + } + if h.MemoSaturated() || h.MemoSaturatedOnce() { + t.Fatal("a nil handle is never saturated") + } + if h.TryLockFor(time.Millisecond) { + t.Fatal("a nil handle cannot be locked") + } + if h.Context() == nil { + t.Fatal("a nil handle should still yield a usable context") + } + if _, _, err := h.PrepareCached("SELECT 1"); err == nil { + t.Fatal("PrepareCached on a nil handle should error, not panic") + } + + // Must not panic. + h.Lock() + h.Unlock() + h.MarkDone() + h.CancelInFlight() + _, release := h.OpContext(0) + release() +} + +func TestWarnOnceIsPerHandlePerKey(t *testing.T) { + h := NewHandle("conn-a", nil, nil, nil) + + if !h.WarnOnce("conn-b") { + t.Fatal("first warning for conn-b should fire") + } + if h.WarnOnce("conn-b") { + t.Fatal("second warning for conn-b should be suppressed") + } + if !h.WarnOnce("conn-c") { + t.Fatal("a different connection should warn independently") + } + + // Per handle, not process-wide: the next subflow invocation warns again. + if !NewHandle("conn-a", nil, nil, nil).WarnOnce("conn-b") { + t.Fatal("a fresh handle must warn again") + } +} + +// TestOpContextTokenKeyed covers the overlap case a single cancel slot gets wrong: B's +// registration would evict A's, and then A's release would clear B's, leaving CancelInFlight +// with nothing to cancel. +func TestOpContextTokenKeyed(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + ctxA, releaseA := h.OpContext(0) + ctxB, releaseB := h.OpContext(0) + defer releaseA() + defer releaseB() + + h.opMu.Lock() + n := len(h.opCancels) + h.opMu.Unlock() + if n != 2 { + t.Fatalf("registered %d cancels, want 2", n) + } + + h.CancelInFlight() + + for name, ctx := range map[string]context.Context{"A": ctxA, "B": ctxB} { + select { + case <-ctx.Done(): + case <-time.After(time.Second): + t.Fatalf("CancelInFlight did not cancel statement context %s", name) + } + } +} + +func TestOpContextReleaseDeregisters(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + _, release := h.OpContext(0) + release() + + h.opMu.Lock() + n := len(h.opCancels) + h.opMu.Unlock() + + if n != 0 { + t.Fatalf("%d cancels still registered after release, want 0", n) + } +} + +// TestOpContextParentIsBaseCtxNotBackground guards the reason OpContext exists: a per-statement +// deadline must be able to interrupt the statement, but must derive from the transaction's own +// context rather than context.Background(). +func TestOpContextParentIsBaseCtxNotBackground(t *testing.T) { + db, _ := newFakeDB(0) + defer db.Close() + + base, cancelBase := context.WithCancel(context.Background()) + tx, err := db.BeginTx(base, nil) + if err != nil { + t.Fatalf("BeginTx: %v", err) + } + h := NewHandle("conn-a", db, tx, base) + + opCtx, release := h.OpContext(0) + defer release() + + cancelBase() + + select { + case <-opCtx.Done(): + case <-time.After(time.Second): + t.Fatal("statement context should derive from the transaction's base context") + } +} + +func TestTryLockForTimesOutRatherThanBlockingForever(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + h.Lock() // simulate an abandoned Eval goroutine still holding the operation lock + defer h.Unlock() + + start := time.Now() + if h.TryLockFor(50 * time.Millisecond) { + t.Fatal("TryLockFor should not have acquired a held lock") + } + if elapsed := time.Since(start); elapsed < 40*time.Millisecond { + t.Fatalf("TryLockFor returned after %v, want it to wait out the timeout", elapsed) + } +} + +func TestTryLockForAcquiresWhenFree(t *testing.T) { + h, _, cleanup := beginFake(t, 0) + defer cleanup() + + if !h.TryLockFor(time.Second) { + t.Fatal("TryLockFor should acquire a free lock") + } + h.Unlock() +} + +// TestConcurrentUseIsRaceFree drives the D3 discipline the way concurrent transition branches +// would. It exists to be run under -race. +func TestConcurrentUseIsRaceFree(t *testing.T) { + h, _, cleanup := beginFake(t, 1) + defer cleanup() + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + h.Lock() + defer h.Unlock() + + if _, _, err := h.PrepareCached("SELECT " + strconv.Itoa(i%3)); err != nil { + t.Errorf("goroutine %d: %v", i, err) + } + h.IsDone() + h.WarnOnce("conn-" + strconv.Itoa(i%2)) + _, release := h.OpContext(0) + release() + }(i) + } + + waited := make(chan struct{}) + go func() { wg.Wait(); close(waited) }() + + select { + case <-waited: + case <-time.After(10 * time.Second): + t.Fatal("concurrent handle use deadlocked") + } +} diff --git a/support/sqltx/manager.go b/support/sqltx/manager.go new file mode 100644 index 0000000..6292294 --- /dev/null +++ b/support/sqltx/manager.go @@ -0,0 +1,79 @@ +package sqltx + +import ( + "context" + "sort" + "strings" + + "github.com/project-flogo/core/support/connection" + "github.com/project-flogo/core/support/log" +) + +// FromManager returns the ambient transaction handle for m, or nil. +// +// It is the ONLY entry point a connector needs, and it is the single home of the +// foreign-connection warning, so all four connectors get identical wording for free. +// +// Safe on a nil context: core/support/test.TestActivityContext.GoContext() returns nil, and so +// does flow's LegacyCtx, and both are on paths connectors already take. +// +// On the non-transactional path this costs one ctx.Value walk, because HasAny short-circuits +// before the manager is inspected at all. +func FromManager(ctx context.Context, m connection.Manager, logger log.Logger) *Handle { + if ctx == nil || m == nil || !HasAny(ctx) { + return nil + } + + id := connection.GetId(m) + if id != "" { + if h := FromContext(ctx, id); h != nil { + if logger != nil && logger.DebugEnabled() { + // The feature's canary. The subflow activity logs the id it enlisted; this logs + // the id it derived. If those two strings ever differ, the whole feature + // silently no-ops with no error anywhere. + logger.Debugf("enlisting in the transaction of the enclosing transactional subflow on connection '%s'", id) + } + return h + } + } + + // Reached only inside a transactional subflow, on a connection that is not the enlisted one. + // Not an error: writing an audit row deliberately outside the transaction is legitimate. But + // it must never be silent. + warnForeign(ctx, m, id, logger) + + return nil +} + +// warnForeign emits the D13 warning at most once per handle per foreign connection. +func warnForeign(ctx context.Context, m connection.Manager, id string, logger log.Logger) { + if logger == nil { + return + } + + enlisted := ConnIDs(ctx) + if len(enlisted) == 0 { + return + } + sort.Strings(enlisted) // stable message and stable dedup anchor + + // Dedup through any ambient handle: per-handle means per subflow invocation, per-key means + // per foreign connection. + anchor := FromContext(ctx, enlisted[0]) + + key := id + if key == "" { + key = "inline:" + m.Type() + } + if !anchor.WarnOnce(key) { + return + } + + enlistedIDs := strings.Join(enlisted, ", ") + if id != "" { + logger.Warnf("connection '%s' is not the connection enlisted in the enclosing transactional subflow (%s); its statements commit independently", id, enlistedIDs) + return + } + + logger.Warnf("this activity uses an inline (non-shared) connection, which is not the connection enlisted in the enclosing transactional subflow (%s); its statements commit independently", enlistedIDs) +} diff --git a/support/sqltx/sqltx.go b/support/sqltx/sqltx.go new file mode 100644 index 0000000..a79cf4d --- /dev/null +++ b/support/sqltx/sqltx.go @@ -0,0 +1,110 @@ +// Package sqltx carries a database/sql transaction from the flow engine to the connector +// activities that must enlist in it (FLOGO-19484). +// +// The registry is ONE context key holding an immutable map[connID]*Handle, replaced +// copy-on-write. Copy-on-write is mandatory: with FLOGO_FLOW_EXECUTE_BRANCHES_CONCURRENTLY=true +// sibling branches read the map simultaneously, so WithHandle must never mutate a map that is +// already visible to another goroutine. +// +// Keying by connection id is what keeps the transaction scoped: an activity only enlists when it +// uses the very connection the transactional subflow declared. Two different connections inside +// one subflow cannot cross-contaminate. +package sqltx + +import "context" + +type ctxKey struct{} + +// registry is immutable once stored in a context. Never mutate a map reached through +// ctx.Value; always copy. +type registry map[string]*Handle + +func fromCtx(ctx context.Context) registry { + if ctx == nil { + return nil + } + r, _ := ctx.Value(ctxKey{}).(registry) + return r +} + +// WithHandle returns a context carrying h for connID. +// +// It is a no-op when h is nil or connID is empty, so a caller can never store a typed nil that +// later reads back as a non-nil interface. +func WithHandle(ctx context.Context, connID string, h *Handle) context.Context { + if ctx == nil { + ctx = context.Background() + } + if h == nil || connID == "" { + return ctx + } + + old := fromCtx(ctx) + next := make(registry, len(old)+1) + for id, existing := range old { + next[id] = existing + } + next[connID] = h + + return context.WithValue(ctx, ctxKey{}, next) +} + +// FromContext returns the handle registered for connID, or nil. Safe on a nil context. +func FromContext(ctx context.Context, connID string) *Handle { + if connID == "" { + return nil + } + return fromCtx(ctx)[connID] +} + +// HasAny reports whether ctx carries any handle at all. This is the fast path every connector +// takes before doing anything else: one ctx.Value walk on the non-transactional path. +func HasAny(ctx context.Context) bool { + return len(fromCtx(ctx)) > 0 +} + +// ConnIDs returns the connection ids ctx carries, unordered. Used by the nested-transaction +// guard and by the D13 warning message. +func ConnIDs(ctx context.Context) []string { + r := fromCtx(ctx) + if len(r) == 0 { + return nil + } + + ids := make([]string, 0, len(r)) + for id := range r { + ids = append(ids, id) + } + + return ids +} + +// Propagate copies src's registry onto dst, preserving dst's own cancellation and deadline. +// +// It returns dst UNCHANGED - identically, with no allocation - when src carries no registry. +// That identity matters: flow's TestGoContextEvalCtxOverride asserts GoContext() returns exactly +// the context it was given, and every non-transactional flow goes through this path. +func Propagate(src, dst context.Context) context.Context { + r := fromCtx(src) + if len(r) == 0 { + return dst + } + if dst == nil { + dst = context.Background() + } + + return context.WithValue(dst, ctxKey{}, r) +} + +// Without strips the registry. Used for detached subflows, which outlive the transaction and +// must never enlist in it. +func Without(ctx context.Context) context.Context { + if ctx == nil { + return context.Background() + } + if !HasAny(ctx) { + return ctx + } + + return context.WithValue(ctx, ctxKey{}, registry(nil)) +} diff --git a/support/sqltx/sqltx_test.go b/support/sqltx/sqltx_test.go new file mode 100644 index 0000000..0e1f71f --- /dev/null +++ b/support/sqltx/sqltx_test.go @@ -0,0 +1,150 @@ +package sqltx + +import ( + "context" + "sort" + "testing" +) + +func TestWithHandleFromContextRoundTrip(t *testing.T) { + h := NewHandle("conn-a", nil, nil, nil) + + ctx := WithHandle(context.Background(), "conn-a", h) + + if got := FromContext(ctx, "conn-a"); got != h { + t.Fatalf("FromContext(conn-a) = %v, want the handle we stored", got) + } + if !HasAny(ctx) { + t.Fatal("HasAny = false, want true") + } +} + +func TestFromContextWrongConnIDReturnsNil(t *testing.T) { + ctx := WithHandle(context.Background(), "conn-a", NewHandle("conn-a", nil, nil, nil)) + + // This is the silent-no-op guard: if a connector derives the wrong id, it must get nil + // rather than somebody else's transaction. + if got := FromContext(ctx, "conn-b"); got != nil { + t.Fatalf("FromContext(conn-b) = %v, want nil", got) + } + if got := FromContext(ctx, ""); got != nil { + t.Fatalf("FromContext(\"\") = %v, want nil", got) + } +} + +func TestNilContextIsSafe(t *testing.T) { + // core/support/test.TestActivityContext.GoContext() and flow's LegacyCtx both return nil, + // and connectors call straight through. None of these may panic. + if FromContext(nil, "conn-a") != nil { + t.Fatal("FromContext(nil, ...) should be nil") + } + if HasAny(nil) { + t.Fatal("HasAny(nil) should be false") + } + if ConnIDs(nil) != nil { + t.Fatal("ConnIDs(nil) should be nil") + } + if got := WithHandle(nil, "conn-a", NewHandle("conn-a", nil, nil, nil)); got == nil { + t.Fatal("WithHandle(nil, ...) should return a usable context") + } + if got := Without(nil); got == nil { + t.Fatal("Without(nil) should return a usable context") + } +} + +func TestWithHandleNilHandleOrEmptyIDIsNoOp(t *testing.T) { + base := context.Background() + + if got := WithHandle(base, "conn-a", nil); HasAny(got) { + t.Fatal("storing a nil handle must not create a registry entry") + } + if got := WithHandle(base, "", NewHandle("", nil, nil, nil)); HasAny(got) { + t.Fatal("storing under an empty connID must not create a registry entry") + } +} + +func TestWithHandleIsCopyOnWrite(t *testing.T) { + a := NewHandle("conn-a", nil, nil, nil) + b := NewHandle("conn-b", nil, nil, nil) + + ctxA := WithHandle(context.Background(), "conn-a", a) + ctxAB := WithHandle(ctxA, "conn-b", b) + + // Concurrent sibling branches read the map simultaneously, so adding to a derived context + // must never mutate the map the parent context still points at. + if FromContext(ctxA, "conn-b") != nil { + t.Fatal("WithHandle mutated the parent context's registry") + } + if FromContext(ctxAB, "conn-a") != a || FromContext(ctxAB, "conn-b") != b { + t.Fatal("derived context lost an entry") + } +} + +func TestConnIDs(t *testing.T) { + ctx := WithHandle(context.Background(), "conn-a", NewHandle("conn-a", nil, nil, nil)) + ctx = WithHandle(ctx, "conn-b", NewHandle("conn-b", nil, nil, nil)) + + ids := ConnIDs(ctx) + sort.Strings(ids) + + if len(ids) != 2 || ids[0] != "conn-a" || ids[1] != "conn-b" { + t.Fatalf("ConnIDs = %v, want [conn-a conn-b]", ids) + } +} + +// TestPropagateIsIdentityWhenSourceHasNoRegistry is load-bearing: flow's +// TestGoContextEvalCtxOverride asserts GoContext() returns *exactly* the context it was given, +// and every non-transactional flow goes through Propagate. If this allocates, that test breaks +// and every flow pays for a feature it does not use. +func TestPropagateIsIdentityWhenSourceHasNoRegistry(t *testing.T) { + dst := context.WithValue(context.Background(), struct{ k string }{"unrelated"}, 1) + + if got := Propagate(context.Background(), dst); got != dst { + t.Fatal("Propagate must return dst identically when src carries no registry") + } +} + +func TestPropagateCopiesRegistryAndKeepsDstCancellation(t *testing.T) { + h := NewHandle("conn-a", nil, nil, nil) + src := WithHandle(context.Background(), "conn-a", h) + + dst, cancel := context.WithCancel(context.Background()) + defer cancel() + + merged := Propagate(src, dst) + + if FromContext(merged, "conn-a") != h { + t.Fatal("Propagate did not copy the registry") + } + + cancel() + select { + case <-merged.Done(): + default: + t.Fatal("Propagate must preserve dst's cancellation") + } +} + +func TestWithoutStripsTheRegistry(t *testing.T) { + ctx := WithHandle(context.Background(), "conn-a", NewHandle("conn-a", nil, nil, nil)) + + stripped := Without(ctx) + + if HasAny(stripped) { + t.Fatal("Without must strip the registry") + } + if FromContext(stripped, "conn-a") != nil { + t.Fatal("Without must make handles unreachable") + } + // A detached subflow must not inherit the transaction; the parent keeps it. + if !HasAny(ctx) { + t.Fatal("Without must not mutate the source context") + } +} + +func TestWithoutIsIdentityWhenNoRegistry(t *testing.T) { + base := context.Background() + if got := Without(base); got != base { + t.Fatal("Without should return the context unchanged when there is no registry") + } +}