Initial distributor implementation

appengine/cmd/dm/mutate/ensure_attempt.go

Index: appengine/cmd/dm/mutate/ensure_attempt.go
diff --git a/appengine/cmd/dm/mutate/ensure_attempt.go b/appengine/cmd/dm/mutate/ensure_attempt.go
index 005c51e9c9f96575b3f360f8eca6af8ccaa62c34..25931c0d2792557af609837def087d9493a45365 100644
--- a/appengine/cmd/dm/mutate/ensure_attempt.go
+++ b/appengine/cmd/dm/mutate/ensure_attempt.go
@@ -21,7 +21,7 @@ type EnsureAttempt struct {
 
 // Root implements tumble.Mutation.
 func (e *EnsureAttempt) Root(c context.Context) *datastore.Key {
-	return datastore.Get(c).KeyForObj(&model.Attempt{ID: *e.ID})
+	return model.AttemptKeyFromID(c, e.ID)
 }
 
 // RollForward implements tumble.Mutation.
@@ -29,14 +29,14 @@ func (e *EnsureAttempt) RollForward(c context.Context) (muts []tumble.Mutation,
 	ds := datastore.Get(c)
 
 	a := model.MakeAttempt(c, e.ID)
-	err = ds.Get(a)
-	if err != datastore.ErrNoSuchEntity {
+	if err = ds.Get(a); err != datastore.ErrNoSuchEntity {
 		return
 	}
 
 	if err = ds.Put(a); err != nil {
 		logging.WithError(err).Errorf(logging.SetField(c, "id", e.ID), "in put")
 	}
+	muts = append(muts, &ScheduleExecution{e.ID})
 	return
 }