txn

Documentation

Overview

Package txn implements support for multi-document transactions.

For details check the following blog post:

http://blog.labix.org/2012/08/22/multi-doc-transactions-for-mongodb

Index

• Constants
• Variables
• func SetChaos(c Chaos)
• func SetDebug(debug bool)
• func SetLogger(l log_Logger)
• type Chaos
• type Op
• type Runner
  • func NewRunner(tc *mgo.Collection) *Runner
  • func (r *Runner) ChangeLog(logc *mgo.Collection)
  • func (r *Runner) PurgeMissing(collections ...string) error
  • func (r *Runner) Resume(id bson.ObjectId) (err error)
  • func (r *Runner) ResumeAll() (err error)
  • func (r *Runner) Run(ops []Op, id bson.ObjectId, info interface{}) (err error)
  • func (r *Runner) SetOptions(opts RunnerOptions)
• type RunnerOptions
  • func DefaultRunnerOptions() RunnerOptions

Constants

const (
	// DocExists and DocMissing may be used on an operation's
	// Assert value to assert that the document with the given
	// Id exists or does not exist, respectively.
	DocExists  = "d+"
	DocMissing = "d-"
)

Variables

var ErrAborted = fmt.Errorf("transaction aborted")

var ErrChaos = fmt.Errorf("interrupted by chaos")

Functions

func SetChaos

func SetChaos(c Chaos)

SetChaos sets the failure injection parameters to c.

func SetDebug

func SetDebug(debug bool)

SetDebug enables the delivery of debug messages to the logger. Only meaningful if a logger is also set.

func SetLogger

func SetLogger(l log_Logger)

SetLogger specifies the *log.Logger object where log messages should be sent to.

Types

type Chaos

type Chaos struct {
	// KillChance is the 0.0 to 1.0 chance that a given checkpoint
	// within the algorithm will raise an interruption that will
	// stop the procedure.
	KillChance float64

	// SlowdownChance is the 0.0 to 1.0 chance that a given checkpoint
	// within the algorithm will be delayed by Slowdown before
	// continuing.
	SlowdownChance float64
	Slowdown       time.Duration

	// If Breakpoint is set, the above settings will only affect the
	// named breakpoint.
	Breakpoint string
}

Chaos holds parameters for the failure injection mechanism.

type Op

type Op struct {
	// C and Id identify the collection and document this operation
	// refers to. Id is matched against the "_id" document field.
	C  string      `bson:"c"`
	Id interface{} `bson:"d"`

	// Assert optionally holds a query document that is used to
	// test the operation document at the time the transaction is
	// going to be applied. The assertions for all operations in
	// a transaction are tested before any changes take place,
	// and the transaction is entirely aborted if any of them
	// fails. This is also the only way to prevent a transaction
	// from being being applied (the transaction continues despite
	// the outcome of Insert, Update, and Remove).
	Assert interface{} `bson:"a,omitempty"`

	// The Insert, Update and Remove fields describe the mutation
	// intended by the operation. At most one of them may be set
	// per operation. If none are set, Assert must be set and the
	// operation becomes a read-only test.
	//
	// Insert holds the document to be inserted at the time the
	// transaction is applied. The Id field will be inserted
	// into the document automatically as its _id field. The
	// transaction will continue even if the document already
	// exists. Use Assert with txn.DocMissing if the insertion is
	// required.
	//
	// Update holds the update document to be applied at the time
	// the transaction is applied. The transaction will continue
	// even if a document with Id is missing. Use Assert to
	// test for the document presence or its contents.
	//
	// Remove indicates whether to remove the document with Id.
	// The transaction continues even if the document doesn't yet
	// exist at the time the transaction is applied. Use Assert
	// with txn.DocExists to make sure it will be removed.
	Insert interface{} `bson:"i,omitempty"`
	Update interface{} `bson:"u,omitempty"`
	Remove bool        `bson:"r,omitempty"`
}

Op represents an operation to a single document that may be applied as part of a transaction with other operations.

type Runner

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

A Runner applies operations as part of a transaction onto any number of collections within a database. See the Run method for details.

func NewRunner

func NewRunner(tc *mgo.Collection) *Runner

NewRunner returns a new transaction runner that uses tc to hold its transactions.

Multiple transaction collections may exist in a single database, but all collections that are touched by operations in a given transaction collection must be handled exclusively by it.

A second collection with the same name of tc but suffixed by ".stash" will be used for implementing the transactional behavior of insert and remove operations.

func (*Runner) ChangeLog

func (r *Runner) ChangeLog(logc *mgo.Collection)

ChangeLog enables logging of changes to the given collection every time a transaction that modifies content is done being applied.

Saved documents are in the format:

{"_id": <txn id>, <collection>: {"d": [<doc id>, ...], "r": [<doc revno>, ...]}}

The document revision is the value of the txn-revno field after the change has been applied. Negative values indicate the document was not present in the collection. Revisions will not change when updates or removes are applied to missing documents or inserts are attempted when the document isn't present.

func (*Runner) PurgeMissing

func (r *Runner) PurgeMissing(collections ...string) error

PurgeMissing removes from collections any state that refers to transaction documents that for whatever reason have been lost from the system (removed by accident or lost in a hard crash, for example).

This method should very rarely be needed, if at all, and should never be used during the normal operation of an application. Its purpose is to put a system that has seen unavoidable corruption back in a working state.

func (*Runner) Resume

func (r *Runner) Resume(id bson.ObjectId) (err error)

Resume resumes the transaction with id. It returns mgo.ErrNotFound if the transaction is not found. Otherwise, it has the same semantics of the Run method after the transaction is inserted.

func (*Runner) ResumeAll

func (r *Runner) ResumeAll() (err error)

ResumeAll resumes all pending transactions. All ErrAborted errors from individual transactions are ignored.

func (*Runner) Run

func (r *Runner) Run(ops []Op, id bson.ObjectId, info interface{}) (err error)

Run creates a new transaction with ops and runs it immediately. The id parameter specifies the transaction id, and may be written down ahead of time to later verify the success of the change and resume it, when the procedure is interrupted for any reason. If empty, a random id will be generated. The info parameter, if not nil, is included under the "i" field of the transaction document.

Operations across documents are not atomically applied, but are guaranteed to be eventually all applied in the order provided or all aborted, as long as the affected documents are only modified through transactions. If documents are simultaneously modified by transactions and out of transactions the behavior is undefined.

If Run returns no errors, all operations were applied successfully. If it returns ErrAborted, one or more operations can't be applied and the transaction was entirely aborted with no changes performed. Otherwise, if the transaction is interrupted while running for any reason, it may be resumed explicitly or by attempting to apply another transaction on any of the documents targeted by ops, as long as the interruption was made after the transaction document itself was inserted. Run Resume with the obtained transaction id to confirm whether the transaction was applied or not.

Any number of transactions may be run concurrently, with one runner or many.

func (*Runner) SetOptions

func (r *Runner) SetOptions(opts RunnerOptions)

SetOptions allows people to change some of the internal behavior of a Runner.

type RunnerOptions

type RunnerOptions struct {
	// MaxTxnQueueLength is a way to limit bad behavior. Many operations on
	// transaction queues are O(N^2), and transaction queues growing too large
	// are usually indicative of a bug in behavior. This should be larger
	// than the maximum number of concurrent operations to a single document.
	// Normal operations are likely to only ever hit 10 or so, we use a default
	// maximum length of 1000.
	MaxTxnQueueLength int

	// AssertionCleanupLength is the length of a txn-queue that we will start
	// cleaning up even if this is only an assertion against the document (not
	// otherwise modifying the document).
	AssertionCleanupLength int
}

RunnerOptions encapsulates ways you can tweak transaction Runner behavior.

func DefaultRunnerOptions

func DefaultRunnerOptions() RunnerOptions

DefaultRunnerOptions defines default behavior for a Runner. Users can use the DefaultRunnerOptions to only override specific behavior.