func (f *flusher) prepare(t *transaction, force bool) (revnos []int64, err error) {
	if t.State != tpreparing {
		return f.rescan(t, force)
	}
	f.debugf("Preparing %s", t)
	dkeys := t.docKeys()
	sort.Sort(dkeys)
	revno := make(map[docKey]int64)
	info := txnInfo{}
	tt := tokenFor(t)
NextDoc:
	for _, dkey := range dkeys {
		change := mgo.Change{
			Update:    bson.D{{"$addToSet", bson.D{{"txn-queue", tt}}}},
			ReturnNew: true,
		}
		c := f.tc.Database.C(dkey.C)
		cquery := c.FindId(dkey.Id).Select(txnFields)
	RetryDoc:
		change.Upsert = false
		chaos("")
		if _, err := cquery.Apply(change, &info); err == nil {
			if info.Remove == "" {
				revno[dkey] = info.Revno
				f.queue[dkey] = info.Queue
				f.debugf("[A] Prepared document %v with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				continue NextDoc
			} else {
				if err := f.loadAndApply(info.Remove); err != nil {
					return nil, err
				}
				goto RetryDoc
			}
		} else if err != mgo.ErrNotFound {
			return nil, err
		}
		change.Upsert = true
		chaos("")
		_, err := f.sc.FindId(dkey).Apply(change, &info)
		if err != nil {
			return nil, err
		}
		if info.Insert != "" {
			if err := f.loadAndApply(info.Insert); err != nil {
				return nil, err
			}
			goto RetryDoc
		}
		docFound := false
		stashFound := false
		if err = c.FindId(dkey.Id).Select(txnFields).One(&info); err == nil {
			docFound = true
		} else if err != mgo.ErrNotFound {
			return nil, err
		} else if err = f.sc.FindId(dkey).One(&info); err == nil {
			stashFound = true
			if info.Revno == 0 {
				info.Revno = -1
			}
		} else if err != mgo.ErrNotFound {
			return nil, err
		}
		if docFound && info.Remove == "" || stashFound && info.Insert == "" {
			for _, dtt := range info.Queue {
				if dtt != tt {
					continue
				}
				if stashFound {
					f.debugf("[B] Prepared document %v on stash with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				} else {
					f.debugf("[B] Prepared document %v with revno %d and queue: %v", dkey, info.Revno, info.Queue)
				}
				revno[dkey] = info.Revno
				f.queue[dkey] = info.Queue
				continue NextDoc
			}
		}
		f.unstashToken(tt, dkey)
		goto RetryDoc
	}
	nonce := tt.nonce()
	qdoc := bson.D{{"_id", t.Id}, {"s", tpreparing}}
	udoc := bson.D{{"$set", bson.D{{"s", tprepared}, {"n", nonce}}}}
	chaos("set-prepared")
	err = f.tc.Update(qdoc, udoc)
	if err == nil {
		t.State = tprepared
		t.Nonce = nonce
	} else if err == mgo.ErrNotFound {
		f.debugf("Can't save nonce of %s: LOST RACE", tt)
		if err := f.reload(t); err != nil {
			return nil, err
		} else if t.State == tpreparing {
			panic("can't save nonce yet transaction is still preparing")
		} else if t.State != tprepared {
			return t.Revnos, nil
		}
		tt = t.token()
	} else if err != nil {
		return nil, err
	}
	prereqs, found := f.hasPreReqs(tt, dkeys)
	if !found {
		return f.rescan(t, force)
	} else if prereqs && !force {
		f.debugf("Prepared queue with %s [has prereqs & not forced].", tt)
		return nil, errPreReqs
	}
	for _, op := range t.Ops {
		dkey := op.docKey()
		revnos = append(revnos, revno[dkey])
		drevno := revno[dkey]
		switch {
		case op.Insert != nil && drevno < 0:
			revno[dkey] = -drevno+1
		case op.Update != nil && drevno >= 0:
			revno[dkey] = drevno+1
		case op.Remove && drevno >= 0:
			revno[dkey] = -drevno-1
		}
	}
	if !prereqs {
		f.debugf("Prepared queue with %s [no prereqs]. Revnos: %v", tt, revnos)
	} else {
		f.debugf("Prepared queue with %s [forced] Revnos: %v", tt, revnos)
	}
	return revnos, nil
}
prepare injects t's id onto txn-queue for all affected documents
and collects the current txn-queue and txn-revno values during
the process. If the prepared txn-queue indicates that there are
pre-requisite transactions to be applied and the force parameter
is false, errPreReqs will be returned. Otherwise, the current
tip revision numbers for all the documents are returned.
package txn
import (
	"fmt"
	"labix.org/v2/mgo"
	"labix.org/v2/mgo/bson"
	"sort"
)
func flush(r *Runner, t *transaction) error {
	f := &flusher{
		Runner:   r,
		goal:     t,
		goalKeys: make(map[docKey]bool),
		queue:    make(map[docKey][]token),
		debugId:  debugPrefix(),
	}
	for _, dkey := range f.goal.docKeys() {
		f.goalKeys[dkey] = true
	}
	return f.run()
}
type flusher struct {
	*Runner
	goal     *transaction
	goalKeys map[docKey]bool
	queue    map[docKey][]token
	debugId  string
}
func (f *flusher) run() (err error) {
	if chaosEnabled {
		defer f.handleChaos(&err)
	}
	f.debugf("Processing %s", f.goal)
	seen := make(map[bson.ObjectId]*transaction)
	if err := f.recurse(f.goal, seen); err != nil {
		return err
	}
	if f.goal.done() {
		return nil
	}
	successors := make(map[bson.ObjectId][]bson.ObjectId)
	ready := true
	for _, dqueue := range f.queue {
	NextPair:
		for i := 0; i < len(dqueue); i++ {
			pred := dqueue[i]
			predid := pred.id()
			predt := seen[predid]
			if predt == nil || predt.Nonce != pred.nonce() {
				continue
			}
			predsuccids, ok := successors[predid]
			if !ok {
				successors[predid] = nil
			}
			for j := i + 1; j < len(dqueue); j++ {
				succ := dqueue[j]
				succid := succ.id()
				succt := seen[succid]
				if succt == nil || succt.Nonce != succ.nonce() {
					continue
				}
				if _, ok := successors[succid]; !ok {
					successors[succid] = nil
				}
				i = j - 1
				for _, predsuccid := range predsuccids {
					if predsuccid == succid {
						continue NextPair
					}
				}
				successors[predid] = append(predsuccids, succid)
				if succid == f.goal.Id {
					ready = false
				}
				continue NextPair
			}
		}
	}
	f.debugf("Queues: %v", f.queue)
	f.debugf("Successors: %v", successors)
	if ready {
		f.debugf("Goal %s has no real pre-requisites", f.goal)
		return f.advance(f.goal, nil, true)
	}
	sorted := tarjanSort(successors)
	if debugEnabled {
		f.debugf("Tarjan output: %v", sorted)
	}
	pull := make(map[bson.ObjectId]*transaction)
	for i := len(sorted) - 1; i >= 0; i-- {
		scc := sorted[i]
		f.debugf("Flushing %v", scc)
		if len(scc) == 1 {
			pull[scc[0]] = seen[scc[0]]
		}
		for _, id := range scc {
			if err := f.advance(seen[id], pull, true); err != nil {
				return err
			}
		}
		if len(scc) > 1 {
			for _, id := range scc {
				pull[id] = seen[id]
			}
		}
	}
	return nil
}
func (f *flusher) recurse(t *transaction, seen map[bson.ObjectId]*transaction) error {
	seen[t.Id] = t
	err := f.advance(t, nil, false)
	if err != errPreReqs {
		return err
	}
	for _, dkey := range t.docKeys() {
		for _, dtt := range f.queue[dkey] {
			id := dtt.id()
			if seen[id] != nil {
				continue
			}
			qt, err := f.load(id)
			if err != nil {
				return err
			}
			err = f.recurse(qt, seen)
			if err != nil {
				return err
			}
		}
	}
	return nil
}
func (f *flusher) advance(t *transaction, pull map[bson.ObjectId]*transaction, force bool) error {
	for {
		switch t.State {
		case tpreparing, tprepared:
			revnos, err := f.prepare(t, force)
			if err != nil {
				return err
			}
			if t.State != tprepared {
				continue
			}
			if err = f.assert(t, revnos, pull); err != nil {
				return err
			}
			if t.State != tprepared {
				continue
			}
			if err = f.checkpoint(t, revnos); err != nil {
				return err
			}
		case tapplying:
			return f.apply(t, pull)
		case taborting:
			return f.abortOrReload(t, nil, pull)
		case tapplied, taborted:
			return nil
		default:
			panic(fmt.Errorf("transaction in unknown state: %q", t.State))
		}
	}
	panic("unreachable")
}
type stash string
const (
	stashStable stash = ""
	stashInsert stash = "insert"
	stashRemove stash = "remove"
)
type txnInfo struct {
	Queue  []token       `bson:"txn-queue"`
	Revno  int64         `bson:"txn-revno,omitempty"`
	Insert bson.ObjectId `bson:"txn-insert,omitempty"`
	Remove bson.ObjectId `bson:"txn-remove,omitempty"`
}
type stashState string
const (
	stashNew       stashState = ""
	stashInserting stashState = "inserting"
)
var txnFields = bson.D{{"txn-queue", 1}, {"txn-revno", 1}, {"txn-remove", 1}, {"txn-insert", 1}}
var errPreReqs = fmt.Errorf("transaction has pre-requisites and force is false")