state;apiserver: Fix a race - lp bug #1067979

state/state.go

 // Copyright 2012, 2013 Canonical Ltd.
 // Licensed under the AGPLv3, see LICENCE file for details.
 
 // The state package enables reading, observing, and changing
 // the state stored in MongoDB of a whole environment
 // managed by juju.
 package state
 
 import (
         "fmt"
         "net/url"
         "regexp"
         "sort"
         "strconv"
         "strings"
         "sync"
 
+        "github.com/loggo/loggo"
         "labix.org/v2/mgo"
         "labix.org/v2/mgo/bson"
         "labix.org/v2/mgo/txn"
-        "launchpad.net/loggo"
 
         "launchpad.net/juju-core/charm"
         "launchpad.net/juju-core/constraints"
         "launchpad.net/juju-core/environs/config"
         "launchpad.net/juju-core/errors"
         "launchpad.net/juju-core/names"
         "launchpad.net/juju-core/state/multiwatcher"
         "launchpad.net/juju-core/state/presence"
         "launchpad.net/juju-core/state/watcher"
         "launchpad.net/juju-core/utils"
@@ skipping 431 matching lines @@
                         BundleSha256: bundleSha256,
                 }
                 err = st.charms.Insert(cdoc)
                 if err != nil {
                         return nil, fmt.Errorf("cannot add charm %q: %v", curl, err)
                 }
                 return newCharm(st, cdoc)
         } else if err != nil {
                 return nil, err
         }
-        return st.updateCharmDoc(ch, curl, bundleURL, bundleSha256, StillPlaceholder)
+        return st.updateCharmDoc(ch, curl, bundleURL, bundleSha256, stillPlaceholder)
 }
 
 // Charm returns the charm with the given URL. Charms pending upload
 // to storage and placeholders are never returned.
 func (st *State) Charm(curl *charm.URL) (*Charm, error) {
         cdoc := &charmDoc{}
-        what := D{{"_id", curl}, {"placeholder", false}, {"pendingupload", false}}
+        what := D{
+                {"_id", curl},
+                {"placeholder", D{{"$ne", true}}},
+                {"pendingupload", D{{"$ne", true}}},
+        }
         err := st.charms.Find(what).One(&cdoc)
         if err == mgo.ErrNotFound {
                 return nil, errors.NotFoundf("charm %q", curl)
         }
         if err != nil {
                 return nil, fmt.Errorf("cannot get charm %q: %v", curl, err)
         }
         if err := cdoc.Meta.Check(); err != nil {
                 return nil, fmt.Errorf("malformed charm metadata found in state: %v", err)
         }
@@ skipping 105 matching lines @@
         if curl.Revision < 0 {
                 return nil, fmt.Errorf("expected charm URL with revision, got %q", curl)
         }
 
         var (
                 uploadedCharm charmDoc
                 err           error
         )
         for attempt := 0; attempt < 3; attempt++ {
                 // Find an uploaded or pending charm with the given exact curl.
-                err = st.charms.Find(D{{"_id", curl}, {"placeholder", false}}).One(&uploadedCharm)
+                err = st.charms.FindId(curl).One(&uploadedCharm)
                 if err != nil && err != mgo.ErrNotFound {
                         return nil, err
-                } else if err == nil {
-                        // The charm exists and it's either uploaded or not.
-                        // In either case, we just return what we got.
-                        break
-                }
-
-                // No charm found, so create it now as pending.
-                uploadedCharm = charmDoc{
-                        URL:           curl,
-                        PendingUpload: true,
-                }
-                ops := []txn.Op{{
-                        C:      st.charms.Name,
-                        Id:     curl,
-                        Assert: txn.DocMissing,
-                        Insert: uploadedCharm,
-                }}
+                } else if err == nil && !uploadedCharm.Placeholder {
+                        // The charm exists and it's either uploaded or still
+                        // pending, but it's not a placeholder. In any case, we
+                        // just return what we got.
+                        return newCharm(st, &uploadedCharm)
+                } else if err == mgo.ErrNotFound {
+                        // Prepare the pending charm document for insertion.
+                        uploadedCharm = charmDoc{
+                                URL:           curl,
+                                PendingUpload: true,
+                                Placeholder:   false,
+                        }
+                }
+
+                var ops []txn.Op
+                if uploadedCharm.Placeholder {
+                        // Convert the placeholder to a pending charm, while
+                        // asserting the fields updated after an upload have not
+                        // changed yet.
+                        ops = []txn.Op{{
+                                C:  st.charms.Name,
+                                Id: curl,
+                                Assert: D{
+                                        {"bundlesha256", ""},
+                                        {"pendingupload", false},
+                                        {"placeholder", true},
+                                },
+                                Update: D{{"$set", D{
+                                        {"pendingupload", true},
+                                        {"placeholder", false},
+                                }}},
+                        }}
+                        // Update the fields of the document we're returning.
+                        uploadedCharm.PendingUpload = true
+                        uploadedCharm.Placeholder = false
+                } else {
+                        // No charm document with this curl yet, insert it.
+                        ops = []txn.Op{{
+                                C:      st.charms.Name,
+                                Id:     curl,
+                                Assert: txn.DocMissing,
+                                Insert: uploadedCharm,
+                        }}
+                }
+
                 // Run the transaction, and retry on abort.
-                if err = st.runTransaction(ops); err == txn.ErrAborted {
+                err = st.runTransaction(ops)
+                if err == txn.ErrAborted {
                         continue
                 } else if err != nil {
                         return nil, err
-                }
-                break
-        }
-        if err != nil {
-                return nil, ErrExcessiveContention
-        }
-        return newCharm(st, &uploadedCharm)
+                } else if err == nil {
+                        return newCharm(st, &uploadedCharm)
+                }
+        }
+        return nil, ErrExcessiveContention
 }
 
 var (
-        StillPending     = D{{"pendingupload", true}}
-        StillPlaceholder = D{{"placeholder", true}}
+        stillPending     = D{{"pendingupload", true}}
+        stillPlaceholder = D{{"placeholder", true}}
 )
 
 // AddStoreCharmPlaceholder creates a charm document in state for the given charm URL which
 // must reference a charm from the store. The charm document is marked as a placeholder which
 // means that if the charm is to be deployed, it will need to first be uploaded to env storage.
 func (st *State) AddStoreCharmPlaceholder(curl *charm.URL) (err error) {
         // Perform sanity checks first.
         if curl.Schema != "cs" {
                 return fmt.Errorf("expected charm URL with cs schema, got %q", curl)
         }
@@ skipping 57 matching lines @@
                 return nil, err
         }
         var ops []txn.Op
         for _, doc := range docs {
                 if doc.URL.Revision >= curl.Revision {
                         continue
                 }
                 ops = append(ops, txn.Op{
                         C:      st.charms.Name,
                         Id:     doc.URL.String(),
-                        Assert: StillPlaceholder,
+                        Assert: stillPlaceholder,
                         Remove: true,
                 })
         }
         return ops, nil
 }
+
+// ErrCharmAlreadyUploaded is returned by UpdateUploadedCharm() when
+// the given charm is already uploaded and marked as not pending in
+// state.
+type ErrCharmAlreadyUploaded struct {
+        curl *charm.URL
+}
+
+func (e *ErrCharmAlreadyUploaded) Error() string {
+        return fmt.Sprintf("charm %q already uploaded", e.curl)
+}
+
+// IsCharmAlreadyUploadedError returns if the given error is
+// ErrCharmAlreadyUploaded.
+func IsCharmAlreadyUploadedError(err interface{}) bool {
+        if err == nil {
+                return false
+        }
+        _, ok := err.(*ErrCharmAlreadyUploaded)
+        return ok
+}
+
+// ErrCharmRevisionAlreadyModified is returned when a pending or
+// placeholder charm is no longer pending or a placeholder, signaling
+// the charm is available in state with its full information.
+var ErrCharmRevisionAlreadyModified = fmt.Errorf("charm revision already modified")
 
 // UpdateUploadedCharm marks the given charm URL as uploaded and
 // updates the rest of its data, returning it as *state.Charm.
 func (st *State) UpdateUploadedCharm(ch charm.Charm, curl *charm.URL, bundleURL *url.URL, bundleSha256 string) (*Charm, error) {
         doc := &charmDoc{}
         err := st.charms.FindId(curl).One(&doc)
         if err == mgo.ErrNotFound {
                 return nil, errors.NotFoundf("charm %q", curl)
         }
         if err != nil {
                 return nil, err
         }
         if !doc.PendingUpload {
-                return nil, fmt.Errorf("charm %q already uploaded", curl)
-        }
-
-        return st.updateCharmDoc(ch, curl, bundleURL, bundleSha256, StillPending)
-}
-
-var ErrCharmRevisionAlreadyModified = fmt.Errorf("charm revision already modified")
+                return nil, &ErrCharmAlreadyUploaded{curl}
+        }
+
+        return st.updateCharmDoc(ch, curl, bundleURL, bundleSha256, stillPending)
+}
 
 // updateCharmDoc updates the charm with specified URL with the given
 // data, and resets the placeholder and pendingupdate flags.  If the
 // charm is no longer a placeholder or pending (depending on preReq),
 // it returns ErrCharmRevisionAlreadyModified.
 func (st *State) updateCharmDoc(
         ch charm.Charm, curl *charm.URL, bundleURL *url.URL, bundleSha256 string, preReq interface{}) (*Charm, error) {
 
         updateFields := D{{"$set", D{
                 {"meta", ch.Meta()},
@@ skipping 553 matching lines @@
 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
 }