Retracted: Withdrawn; the module proxy serves these against commits that no longer exist here.
Documentation
¶
Overview ¶
Package twopc coordinates two-phase commit across sinks.
The guarantee is that either every participant applied a batch or none did. Everything here exists to hold that across a crash, because a coordinator that only works while the process is alive is not a coordinator — it is an ordering convention.
Semantics ¶
Presumed abort. A transaction is committed on recovery only if the log says the decision was reached; anything else aborts. Committing on doubt would apply a batch the coordinator never agreed to.
The hazard this package is mostly about ¶
A participant that has prepared is *in doubt*: it holds its changes and its locks until someone resolves it. On PostgreSQL a prepared transaction also blocks VACUUM for the whole cluster and survives restarts — an unresolved one is an outage in waiting, not an untidy row. So:
- nothing is prepared before the intent is durable (Run refuses otherwise);
- Recover resolves in-doubt transactions on every start; and
- Reap rolls back anything that has been prepared for too long.
Read the "Operational hazard" section of README.md before enabling this.
Index ¶
- Constants
- Variables
- type Coordinator
- func (c *Coordinator) InDoubt(ctx context.Context) (int, error)
- func (c *Coordinator) Reap(ctx context.Context, sinks map[string]hermod.TwoPhaseCommit) (int, error)
- func (c *Coordinator) Recover(ctx context.Context, sinks map[string]hermod.TwoPhaseCommit) error
- func (c *Coordinator) Run(ctx context.Context, participants []Participant, ...) error
- func (c *Coordinator) WorkflowID() string
- type IndexedStore
- func (s *IndexedStore) Delete(ctx context.Context, key string) error
- func (s *IndexedStore) Get(ctx context.Context, key string) ([]byte, error)
- func (s *IndexedStore) List(ctx context.Context, prefix string) (map[string][]byte, error)
- func (s *IndexedStore) Set(ctx context.Context, key string, value []byte) error
- type Options
- type Participant
- type Store
Constants ¶
const DefaultMaxPreparedAge = 15 * time.Minute
DefaultMaxPreparedAge bounds how long a transaction may sit in doubt before Reap rolls it back. Generous enough to survive a slow commit round, short enough that a forgotten prepared transaction does not quietly wreck autovacuum.
Variables ¶
var ErrNotFound = errors.New("twopc: record not found")
ErrNotFound is what a Store returns for a missing key.
Functions ¶
This section is empty.
Types ¶
type Coordinator ¶
type Coordinator struct {
// contains filtered or unexported fields
}
Coordinator drives two-phase commit over a set of participants.
func (*Coordinator) InDoubt ¶
func (c *Coordinator) InDoubt(ctx context.Context) (int, error)
InDoubt reports the transactions currently awaiting resolution. Intended for operators and metrics: a number that does not return to zero means locks are being held.
func (*Coordinator) Reap ¶
func (c *Coordinator) Reap(ctx context.Context, sinks map[string]hermod.TwoPhaseCommit) (int, error)
Reap rolls back transactions that have been prepared for longer than MaxPreparedAge, returning how many it resolved.
This is the safety valve for the PostgreSQL hazard: a prepared transaction holds its locks and blocks VACUUM cluster-wide until someone resolves it. A coordinator that dies mid-round and never comes back would otherwise leave that forever. Call it periodically.
It only ever rolls back. Committing on a timer would apply a batch nobody decided to commit.
func (*Coordinator) Recover ¶
func (c *Coordinator) Recover(ctx context.Context, sinks map[string]hermod.TwoPhaseCommit) error
Recover resolves every transaction left in doubt by a previous run. Call it once at start-up, before the pipeline writes anything.
sinks maps participant ID to the live sink. A record naming a participant that is not present cannot be resolved; Recover reports that and leaves the record alone rather than deleting evidence of a transaction that still holds locks.
func (*Coordinator) Run ¶
func (c *Coordinator) Run(ctx context.Context, participants []Participant, work func(context.Context) error) error
Run executes work inside a distributed transaction across participants.
work performs the actual writes; it runs *before* the vote, because a participant can only prepare changes it already has. If work fails, nothing is prepared and every participant is rolled back.
Run is safe to call concurrently but serialises internally.
func (*Coordinator) WorkflowID ¶
func (c *Coordinator) WorkflowID() string
WorkflowID reports the workflow this coordinator's transaction log is scoped to. It labels the metrics the reaper publishes, so an in-doubt transaction can be traced back to the workflow that left it.
type IndexedStore ¶
type IndexedStore struct {
// contains filtered or unexported fields
}
IndexedStore adapts a plain hermod.StateStore into a twopc.Store.
The base interface is Get/Set/Delete with no enumeration, but recovery has to answer "what was in flight when we died?" — so this keeps a separate index key listing the live record keys, and List reads through it.
The index is the weak point and worth being explicit about: it is a second write, so a crash between writing a record and indexing it leaves a record that List cannot see. The order below is chosen so that failure is the safe one — index first, then the record. An index entry with no record is harmless (List skips it and prunes it); a record with no index entry would be invisible to recovery, which is the outcome that strands prepared transactions.
A backend that can enumerate natively should implement Store directly and skip this.
func NewIndexedStore ¶
func NewIndexedStore(base hermod.StateStore, indexKey string) *IndexedStore
NewIndexedStore wraps base. indexKey must be unique per coordinator scope.
func (*IndexedStore) Delete ¶
func (s *IndexedStore) Delete(ctx context.Context, key string) error
Delete removes the record first and the index entry second. The reverse order could drop the index entry and then fail, leaving an orphaned record that List never returns.
type Options ¶
type Options struct {
// Store is the durable log. Required: without it there is no recovery, and
// a coordinator that cannot recover is worse than none.
Store Store
// WorkflowID scopes log keys so coordinators do not see each other's
// transactions.
WorkflowID string
// MaxPreparedAge overrides DefaultMaxPreparedAge.
MaxPreparedAge time.Duration
// Logger is optional.
Logger hermod.Logger
// Now is injectable for tests.
Now func() time.Time
}
Options configures a Coordinator.
type Participant ¶
type Participant struct {
ID string
Sink hermod.TwoPhaseCommit
}
Participant is one sink taking part in a transaction. ID must be stable across restarts: recovery matches log records to live sinks by it.
type Store ¶
type Store interface {
Get(ctx context.Context, key string) ([]byte, error)
Set(ctx context.Context, key string, value []byte) error
Delete(ctx context.Context, key string) error
List(ctx context.Context, prefix string) (map[string][]byte, error)
}
Store is the coordinator's durable log.
It is hermod.StateStore plus List, because recovery has to enumerate what was in flight and the base interface cannot. Wrap a plain StateStore with NewIndexedStore to get one.
Source Files