Initial distributor implementation

appengine/cmd/dm/mutate/finish_attempt.go

Index: appengine/cmd/dm/mutate/finish_attempt.go
diff --git a/appengine/cmd/dm/mutate/finish_attempt.go b/appengine/cmd/dm/mutate/finish_attempt.go
index 1b52e8972a4d6cc1b7f06d453a3a0bfeb42a8f9e..b2f4133aef4afb3eb187775c62e305ccabeeb145 100644
--- a/appengine/cmd/dm/mutate/finish_attempt.go
+++ b/appengine/cmd/dm/mutate/finish_attempt.go
@@ -30,7 +30,7 @@ type FinishAttempt struct {
 
 // Root implements tumble.Mutation
 func (f *FinishAttempt) Root(c context.Context) *datastore.Key {
-	return datastore.Get(c).KeyForObj(&model.Attempt{ID: *f.Auth.Id.AttemptID()})
+	return model.AttemptKeyFromID(c, f.Auth.Id.AttemptID())
 }
 
 // RollForward implements tumble.Mutation
@@ -38,16 +38,16 @@ func (f *FinishAttempt) Root(c context.Context) *datastore.Key {
 // This mutation is called directly from FinishAttempt, so we use
 // grpcutil.MaybeLogErr
 func (f *FinishAttempt) RollForward(c context.Context) (muts []tumble.Mutation, err error) {
-	atmpt, _, err := model.InvalidateExecution(c, f.Auth)
+	atmpt, ex, err := model.InvalidateExecution(c, f.Auth)
 	if err != nil {
 		return
 	}
 
-	ds := datastore.Get(c)
+	if err = ResetExecutionTimeout(c, ex); err != nil {
+		return
+	}
 
-	// Executing -> Finished is valid, and we know we're already Executing because
-	// the InvalidateExecution call above asserts that or errors out.
-	atmpt.MustModifyState(c, dm.Attempt_FINISHED)
+	ds := datastore.Get(c)
 
 	atmpt.ResultSize = uint32(len(f.Result))
 	atmpt.ResultExpiration = f.ResultExpiration
@@ -61,10 +61,6 @@ func (f *FinishAttempt) RollForward(c context.Context) (muts []tumble.Mutation,
 	err = grpcutil.MaybeLogErr(c, ds.PutMulti([]interface{}{atmpt, rslt}),
 		codes.Internal, "while trying to PutMulti")
 
-	// TODO(iannucci): also include mutations to generate index entries for
-	// the attempt results.
-	muts = append(muts, &RecordCompletion{For: f.Auth.Id.AttemptID()})
-
 	return
 }