Skip to content

FLOGO-19484: add support/sqltx for ambient DB transaction propagation - #300

Open
awakchau-tibco wants to merge 1 commit into
masterfrom
FLOGO-19484-transactional-subflow
Open

awakchau-tibco wants to merge 1 commit into
masterfrom
FLOGO-19484-transactional-subflow

Conversation

@awakchau-tibco

@awakchau-tibco awakchau-tibco commented Sep 2, 2026

Copy link
Copy Markdown
Collaborator

FLOGO-19484 — Support for DB transactions using a subflow

A subflow can be marked transactional against a chosen SQL connection. The transaction
BEGINs when the subflow starts and is COMMITted or ROLLed BACK automatically from the subflow's
outcome — there is no explicit commit/rollback activity to place in the flow.

It is generic across SQL Server, MySQL, Oracle and PostgreSQL through database/sql. Each
subflow invocation gets its own transaction, parallel requests never share one, and the
transaction reaches the activities inside the subflow through the go context.

How it fits together

  subflow activity          BeginTx on the selected connection, then
  (flow/activity/subflow)   StartTransactionalSubFlow(...) with a commit/rollback finaliser
          │
          ▼
  flow runtime              owns the transaction for the life of the subflow instance,
  (flow/instance)           latches any activity failure, decides commit vs rollback
          │
          ▼  go context
  core/support/sqltx        immutable map[connectionID]*Handle, copy-on-write
          │
          ▼
  connectors                sqltx.FromManager(ctx, mgr, log) -> run on the Tx, or on the pool

The design point worth knowing: the connection manager is a process-wide singleton, so the
transaction cannot live on it. An earlier approach that handed out a "transaction-enabled"
manager was dropped for exactly this reason — it would have leaked one request's transaction into
every other request on the same connection. The context is the only per-invocation carrier
available, which is why sqltx keys off it.

Merge order — this is 1 of 7 PRs

# Repo Scope Blocked on
1 project-flogo/core#300 support/sqltx, connection.GetId - mergeable now
2 tibco/wi-contrib#197 studio settings for the subflow activity - mergeable now
3 project-flogo/flow#224 flow runtime and the activity/subflow module core v1.6.22
4 tibco/wi-mssql#50 SQLServer connector core v1.6.22
5 tibco/wi-mysql#48 MySQL connector core v1.6.22
6 tibco/wi-oracledb#40 Oracle connector core v1.6.22
7 tibco/wi-postgres#35 PostgreSQL connector core v1.6.22

Release sequence:

merge 1 and 2  ->  tag core v1.6.22  ->  merge 3  ->  tag flow v1.6.28  ->  merge 4..7

v1.6.20 and v1.6.21 already exist and do not contain support/sqltx, so the dependent
modules pin v1.6.22 deliberately: an unresolvable version fails loudly, whereas pinning an
existing tag would fail with a confusing "undefined: sqltx" instead.


What this PR adds

support/sqltx — the carrier that lets a transactional subflow hand a live *sql.Tx to the
activities running inside it — plus connection.GetId, which resolves a manager back to its
registered id.

Nothing in core calls this yet. It is the shared vocabulary the flow runtime and the four SQL
connectors are written against, so it has to land first.

+1352 / −0 across 7 files. No existing file changes behaviour; registry.go only gains a
function.

support/sqltx

One context key, holding an immutable map[connID]*Handle, replaced copy-on-write.
A single context can therefore carry several independent transactions (one per connection), and
two flow instances running in parallel never observe each other's registry.

Propagate(src, dst) returns dst identically when src carries no registry, so the
non-transactional path — which is every existing flow — allocates nothing and is unaffected.

Handle serialises access to the single *sql.Tx. database/sql pins a transaction to one
physical connection, so two activities on parallel branches of the same subflow must not use it
concurrently. Three locks, with deliberately separate jobs:

lock guards who takes it
mu the operation itself connectors, via Lock() / TryLockFor()
stmtsMu statement memo, done flag, warn-once set internal
opMu token-keyed in-flight cancel funcs internal

They are separate so CancelInFlight() stays callable while mu is held by the very operation
being cancelled
. Collapsing them into one mutex deadlocks rollback-on-timeout.

PrepareCached prepares on the transaction, never on the pool. This is a correctness
requirement, not a preference: db.Prepare needs a second pooled connection while the
transaction pins the first, so at maxOpenConnection: 1 it deadlocks outright. Preparing on the
Tx removes the pool from the picture entirely.

The memo is bounded at 128 statements and reports saturation exactly once, so a flow that
generates unbounded distinct SQL degrades rather than growing without limit.

connection.GetId

A linear scan over the registry, mirroring the existing IsShared, guarded by recover().

A map[Manager]string index would be O(1) and was rejected: manager implementations are not
required to be comparable, and inserting one with an uncomparable dynamic type panics at app
startup
. A scan over a registry that is written once at startup and then only read is not worth
that risk.

Testing

go test ./support/...support/sqltx and support/connection green.

Covered: registry copy-on-write semantics, concurrent handle use, memo saturation, cancel/finish
races, and the ErrTxFinished path. An in-repo fake driver.Driver means no database is
needed
to run them.

Pre-existing failure, unrelated to this PR

support.TestURLStringToFilePath fails on Windows (\ vs / in the expected path). Verified
identical at origin/master in a detached worktree — it is not affected by this change.

Reviewer notes

  • The only pre-existing file touched is support/connection/registry.go, and only additively.
  • sqltx.Without is currently unused. It is the inverse of WithHandle and exists for a
    suppression case that did not materialise — happy to drop it if you would rather not carry it.

awakchau-tibco added a commit to project-flogo/flow that referenced this pull request Sep 3, 2026
…of core and flow

The module required bare `core v1.6.22` and `flow v1.6.28`. Neither tag exists yet, so the
module could not be resolved, built or tested by anyone -- and `go.sum` could not be generated
at all, because there was nothing to hash.

Worse, it broke consumers in a way that was hard to read. The SQL connectors also require a bare
`core v1.6.22`, and a pseudo-version like `v1.6.22-0.<ts>-<sha>` is a PRE-RELEASE of v1.6.22, so
semver sorts it BELOW the bare tag. MVS therefore discarded any working pseudo-version a consumer
supplied, selected the bare tag, and failed with "unknown revision v1.6.22". The only way out was
a `replace` in every consuming build.

Both are now pinned to the commits on this PR's branch:

  core v1.6.22-0.20260902065001-9e288508778b   project-flogo/core#300
  flow v1.6.28-0.20260902065149-820199433c22   the runtime commit of this PR

flow is pinned at the runtime commit rather than this branch's head deliberately: that commit is
what actually provides StartTransactionalSubFlow, and it does not move as further commits land on
the branch.

The module now builds, vets and tests green with NO replace directives, resolving core and flow
from the remote like any ordinary dependency, and go.sum carries real hashes.

AT RELEASE: once core v1.6.22 and flow v1.6.28 are tagged, replace both pseudo-versions with the
plain tags and re-run `go mod tidy`.
Adds the carrier that lets a transactional subflow hand a live *sql.Tx to the
activities running inside it, plus connection.GetId to resolve a connection
manager back to its registered id.

support/sqltx
  Keeps ONE context key holding an immutable map[connID]*Handle, replaced
  copy-on-write. A single context can therefore carry several independent
  transactions (one per connection), and two flow instances running in parallel
  never observe each other's registry. Propagate() returns the destination
  context unchanged when the source carries no registry, so the non-
  transactional path allocates nothing.

  Handle serialises access to the one *sql.Tx that database/sql pins to a single
  physical connection. Three locks with distinct jobs:
    mu       the operation lock connectors take around a statement
    stmtsMu  guards the prepared-statement memo, done flag and warn-once set
    opMu     guards the token-keyed in-flight cancel funcs
  Splitting them keeps CancelInFlight() usable while mu is held by the very
  operation being cancelled, which is what makes rollback-on-timeout work.

  PrepareCached prepares ON THE TRANSACTION, never on the pool. Preparing on
  *sql.DB would need a second pooled connection while the transaction pins the
  first, which deadlocks outright at maxOpenConnection=1. The memo is bounded at
  128 statements and reports saturation once, so a flow generating unbounded
  distinct SQL degrades instead of growing without limit.

support/connection.GetId
  Linear scan over the registry mirroring the existing IsShared, guarded by
  recover() because manager implementations are not required to be comparable.
  A map[Manager]string index was rejected: it panics at app startup as soon as
  one registered manager has an uncomparable dynamic type.

Tests cover the registry copy-on-write semantics, concurrent handle use, memo
saturation, cancel/finish races and the ErrTxFinished path, using an in-repo
fake driver so no database is required.
@awakchau-tibco
awakchau-tibco force-pushed the FLOGO-19484-transactional-subflow branch from 9e28850 to e7c210a Compare September 11, 2026 08:24
awakchau-tibco added a commit to project-flogo/flow that referenced this pull request Sep 11, 2026
…of core and flow

The module required bare `core v1.6.22` and `flow v1.6.28`. Neither tag exists yet, so the
module could not be resolved, built or tested by anyone -- and `go.sum` could not be generated
at all, because there was nothing to hash.

Worse, it broke consumers in a way that was hard to read. The SQL connectors also require a bare
`core v1.6.22`, and a pseudo-version like `v1.6.22-0.<ts>-<sha>` is a PRE-RELEASE of v1.6.22, so
semver sorts it BELOW the bare tag. MVS therefore discarded any working pseudo-version a consumer
supplied, selected the bare tag, and failed with "unknown revision v1.6.22". The only way out was
a `replace` in every consuming build.

Both are now pinned to the commits on this PR's branch:

  core v1.6.22-0.20260902065001-9e288508778b   project-flogo/core#300
  flow v1.6.28-0.20260902065149-820199433c22   the runtime commit of this PR

flow is pinned at the runtime commit rather than this branch's head deliberately: that commit is
what actually provides StartTransactionalSubFlow, and it does not move as further commits land on
the branch.

The module now builds, vets and tests green with NO replace directives, resolving core and flow
from the remote like any ordinary dependency, and go.sum carries real hashes.

AT RELEASE: once core v1.6.22 and flow v1.6.28 are tagged, replace both pseudo-versions with the
plain tags and re-run `go mod tidy`.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant