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"
@@ skipping 427 matching lines @@
         case names.RelationTagKind:
                 coll = st.relations.Name
         case names.EnvironTagKind:
                 coll = st.environments.Name
         default:
                 return "", "", fmt.Errorf("%q is not a valid collection tag", tag)
         }
         return coll, id, nil
 }
 
-// AddCharm adds the ch charm with curl to the state.  bundleUrl must be
-// set to a URL where the bundle for ch may be downloaded from.
-// On success the newly added charm state is returned.
+// AddCharm adds the ch charm with curl to the state. bundleURL must
+// be set to a URL where the bundle for ch may be downloaded from. On
+// success the newly added charm state is returned.
 func (st *State) AddCharm(ch charm.Charm, curl *charm.URL, bundleURL *url.URL, bundleSha256 string) (stch *Charm, err error) {
-        // The charm may already exist in state as a placeholder, so we check for that situation and update the
-        // existing charm record if necessary, otherwise add a new record.
+        // The charm may already exist in state as a placeholder, so we
+        // check for that situation and update the existing charm record
+        // if necessary, otherwise add a new record.
         var existing charmDoc
         err = st.charms.Find(D{{"_id", curl.String()}, {"placeholder", true}}).One(&existing)
         if err == mgo.ErrNotFound {
                 cdoc := &charmDoc{
                         URL:          curl,
                         Meta:         ch.Meta(),
                         Config:       ch.Config(),
                         BundleURL:    bundleURL,
                         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.
+// 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{}
-        err := st.charms.Find(D{{"_id", curl}, {"pendingupload", false}, {"placeholder", false}}).One(cdoc)
+        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)
         }
         return newCharm(st, cdoc)
 }
 
-// LatestPlaceholderCharm returns the latest charm described by the given URL but which is not yet deployed.
+// LatestPlaceholderCharm returns the latest charm described by the
+// given URL but which is not yet deployed.
 func (st *State) LatestPlaceholderCharm(curl *charm.URL) (*Charm, error) {
         noRevURL := curl.WithRevision(-1)
         curlRegex := "^" + regexp.QuoteMeta(noRevURL.String())
         var docs []charmDoc
         err := st.charms.Find(D{{"_id", D{{"$regex", curlRegex}}}, {"placeholder", true}}).All(&docs)
         if err != nil {
                 return nil, fmt.Errorf("cannot get charm %q: %v", curl, err)
         }
         // Find the highest revision.
         var latest charmDoc
         for _, doc := range docs {
                 if latest.URL == nil || doc.URL.Revision > latest.URL.Revision {
                         latest = doc
                 }
         }
         if latest.URL == nil {
                 return nil, errors.NotFoundf("placeholder charm %q", noRevURL)
         }
         return newCharm(st, &latest)
 }
 
-// PrepareLocalCharmUpload must be called before a charm is uploaded
-// to the provider storage in order to create a charm document in
-// state. It returns a new Charm with no metadata and a unique
-// revision number.
+// PrepareLocalCharmUpload must be called before a local charm is
+// uploaded to the provider storage in order to create a charm
+// document in state. It returns the chosen unique charm URL reserved
+// in state for the charm.
 //
 // The url's schema must be "local" and it must include a revision.
 func (st *State) PrepareLocalCharmUpload(curl *charm.URL) (chosenUrl *charm.URL, err error) {
         // Perform a few sanity checks first.
         if curl.Schema != "local" {
                 return nil, fmt.Errorf("expected charm URL with local schema, got %q", curl)
         }
         if curl.Revision < 0 {
                 return nil, fmt.Errorf("expected charm URL with revision, got %q", curl)
         }
@@ skipping 41 matching lines @@
                         return nil, err
                 }
                 break
         }
         if err != nil {
                 return nil, ErrExcessiveContention
         }
         return chosenUrl, nil
 }
 
-var StillPending = D{{"pendingupload", true}}
-var StillPlaceholder = D{{"placeholder", true}}
+// PrepareStoreCharmUpload must be called before a charm store charm
+// is uploaded to the provider storage in order to create a charm
+// document in state. If a charm with the same URL is already in
+// state, it will be returned as a *state.Charm (is can be still
+// pending or already uploaded). Otherwise, a new charm document is
+// added in state with just the given charm URL and
+// PendingUpload=true, which is then returned as a *state.Charm.
+//
+// The url's schema must be "cs" and it must include a revision.
+func (st *State) PrepareStoreCharmUpload(curl *charm.URL) (*Charm, error) {
+        // Perform a few sanity checks first.
+        if curl.Schema != "cs" {
+                return nil, fmt.Errorf("expected charm URL with cs schema, got %q", curl)
+        }
+        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.FindId(curl).One(&uploadedCharm)
+                if err != nil && err != mgo.ErrNotFound {
+                        return nil, err
+                } 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.
+                        break
+                } 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.
+                        ops = []txn.Op{{
+                                C:      st.charms.Name,
+                                Id:     curl,
+                                Assert: txn.DocExists,

[fwereade, 2014/01/29 16:01:22]

hmm, should be be asserting that it hasn't had archiveurl (etc) filled in? I
think there's a very very cornery corner case in which we read while it's still
a placeholder, but write when it's already uploaded.

That'd be a good transaction-hooks test ;).

[dimitern, 2014/01/29 20:59:13]

Done.

+                                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.
+                err = st.runTransaction(ops)
+                if err == txn.ErrAborted {
+                        continue
+                } else if err != nil {
+                        return nil, err
+                }
+                break
+        }
+        if err != nil {

[fwereade, 2014/01/29 16:01:22]

this makes me twitch a little -- AFAICT the only possible value is
txn.ErrAborted, but that assumption feels very vulnerable to code perturbations
in the loop above. Maybe check for err == nil inside the loop, and return
newCharm() there, so that ErrAborted becomes the *only* way to fall out of the
loop, and just always return ErrExcessiveContention?

[dimitern, 2014/01/29 20:59:13]

Done.

+                return nil, ErrExcessiveContention
+        }
+        return newCharm(st, &uploadedCharm)
+}
+
+var (
+        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)
         }
         if curl.Revision < 0 {
@@ skipping 56 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 nil, &ErrCharmAlreadyUploaded{curl}
         }
 
-        return st.updateCharmDoc(ch, curl, bundleURL, bundleSha256, StillPending)
+        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.
+// 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()},
                 {"config", ch.Config()},
                 {"bundleurl", bundleURL},
                 {"bundlesha256", bundleSha256},
                 {"pendingupload", false},
                 {"placeholder", false},
         }}}
         ops := []txn.Op{{
                 C:      st.charms.Name,
                 Id:     curl,
                 Assert: preReq,
                 Update: updateFields,
         }}
         if err := st.runTransaction(ops); err != nil {
-                return nil, onAbort(err, fmt.Errorf("charm revision already modified"))
+                return nil, onAbort(err, ErrCharmRevisionAlreadyModified)
         }
         return st.Charm(curl)
 }
 
 // addPeerRelationsOps returns the operations necessary to add the
 // specified service peer relations to the state.
 func (st *State) addPeerRelationsOps(serviceName string, peers map[string]charm.Relation) ([]txn.Op, error) {
         var ops []txn.Op
         for _, rel := range peers {
                 relId, err := st.sequence("relation")
@@ skipping 529 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
 }