state: testable transactions

state/state.go

 // The state package enables reading, observing, and changing
 // the state stored in MongoDB of a whole environment
 // managed by juju.
 package state
 
 import (
         "fmt"
         "labix.org/v2/mgo"
         "labix.org/v2/mgo/bson"
         "labix.org/v2/mgo/txn"
@@ skipping 112 matching lines @@
         settings       *mgo.Collection
         settingsrefs   *mgo.Collection
         constraints    *mgo.Collection
         units          *mgo.Collection
         users          *mgo.Collection
         presence       *mgo.Collection
         cleanups       *mgo.Collection
         annotations    *mgo.Collection
         statuses       *mgo.Collection
         runner         *txn.Runner
+        txnHooks       chan ([]func())
         watcher        *watcher.Watcher
         pwatcher       *presence.Watcher
         // mu guards allManager.
         mu         sync.Mutex
         allManager *multiwatcher.StoreManager
 }
 
+// runTxn runs the supplied operations as a single mgo/txn transaction, and
+// includes a mechanism whereby tests can use SetTxnHooks to induce arbitrary
+// state mutations before and after particular transactions are attempted.
+func (st *State) runTxn(ops []txn.Op) error {

[thumper, 2013/05/09 16:13:58]

I have to say that this function name has one of my pet hates.  Any reason why
this can't be runTransaction?  Can we start using full words?

[fwereade, 2013/05/15 09:29:22]

I'm fine with full names in general. Keep hassling me when I forget, it'll
become second nature again one day ;). 

+        txnHooks := <-st.txnHooks
+        st.txnHooks <- nil
+        switch len(txnHooks) {
+        default:
+                defer func() {
+                        if txnHooks[1] != nil {
+                                txnHooks[1]()
+                        }
+                        if <-st.txnHooks != nil {
+                                panic("concurrent use of transaction hooks")

[dimitern, 2013/05/03 15:06:29]

this is to safeguard against txns running in goroutines, right?

[fwereade, 2013/05/15 09:29:22]

Yeah -- concurrent use of this mechanism is not sane, and I'm fine with
panicking the odd insane test. Should be perfectly safe otherwise. (I don't
think it'd be sane to implement this without the buffered txnHooks channel,
though -- it has to be sane in concurrent usage when there aren't any hooks in
play.)

+                        }
+                        st.txnHooks <- txnHooks[2:]
+                }()
+                fallthrough
+        case 1:
+                if txnHooks[0] != nil {
+                        txnHooks[0]()
+                }
+        case 0:
+        }
+        return st.runner.Run(ops, "", nil)
+}
+
 func (st *State) Watch() *multiwatcher.Watcher {
         st.mu.Lock()
         if st.allManager == nil {
                 st.allManager = multiwatcher.NewStoreManager(newAllWatcherStateBacking(st))
         }
         st.mu.Unlock()
         return multiwatcher.NewWatcher(st.allManager)
 }
 
 func (st *State) EnvironConfig() (*config.Config, error) {
@@ skipping 109 matching lines @@
                 Series:     series,
                 InstanceId: instanceId,
                 Nonce:      nonce,
                 Jobs:       jobs,
         }
         mdoc, ops, err := st.addMachineOps(mdoc, cons)
         if err != nil {
                 return nil, err
         }
 
-        err = st.runner.Run(ops, "", nil)
+        err = st.runTxn(ops)
         if err != nil {
                 return nil, err
         }
         // Refresh to pick the txn-revno.
         m = newMachine(st, mdoc)
         if err = m.Refresh(); err != nil {
                 return nil, err
         }
         return m, nil
 }
@@ skipping 289 matching lines @@
         // Collect peer relation addition operations.
         peerOps, err := st.addPeerRelationsOps(name, peers)
         if err != nil {
                 return nil, err
         }
         ops = append(ops, peerOps...)
 
         // Run the transaction; happily, there's never any reason to retry,
         // because all the possible failed assertions imply that the service
         // already exists.
-        if err := st.runner.Run(ops, "", nil); err == txn.ErrAborted {
+        if err := st.runTxn(ops); err == txn.ErrAborted {
                 return nil, fmt.Errorf("service already exists")
         } else if err != nil {
                 return nil, err
         }
         // Refresh to pick the txn-revno.
         if err = svc.Refresh(); err != nil {
                 return nil, err
         }
         return svc, nil
 }
@@ skipping 221 matching lines @@
                         Endpoints: eps,
                         Life:      Alive,
                 }
                 ops = append(ops, txn.Op{
                         C:      st.relations.Name,
                         Id:     doc.Key,
                         Assert: txn.DocMissing,
                         Insert: doc,
                 })
                 // Run the transaction, and retry on abort.
-                if err = st.runner.Run(ops, "", nil); err == txn.ErrAborted {
+                if err = st.runTxn(ops); err == txn.ErrAborted {
                         continue
                 } else if err != nil {
                         return nil, err
                 }
                 return &Relation{st, *doc}, nil
         }
         return nil, ErrExcessiveContention
 }
 
 // EndpointsRelation returns the existing relation with the given endpoints.
@@ skipping 204 matching lines @@
                         // delete directly.
                         if _, err := c.RemoveAll(sel); err != nil {
                                 return fmt.Errorf("cannot remove documents marked for cleanup: %v", err)
                         }
                 }
                 ops := []txn.Op{{
                         C:      st.cleanups.Name,
                         Id:     doc.Id,
                         Remove: true,
                 }}
-                if err := st.runner.Run(ops, "", nil); err != nil {
+                if err := st.runTxn(ops); err != nil {
                         return fmt.Errorf("cannot remove empty cleanup document: %v", err)
                 }
         }
         if err := iter.Err(); err != nil {
                 return fmt.Errorf("cannot read cleanup document: %v", err)
         }
         return nil
 }
 
 var tagPrefix = map[byte]string{
         'm': "machine-",
         's': "service-",
         'u': "unit-",
         'e': "environment-",
 }
 
 func tagForGlobalKey(key string) (string, bool) {
         if len(key) < 3 || key[1] != '#' {
                 return "", false
         }
         p, ok := tagPrefix[key[0]]
         if !ok {
                 return "", false
         }
         return p + key[2:], true
 }