Initial distributor implementation

appengine/cmd/dm/mutate/add_deps_test.go

Index: appengine/cmd/dm/mutate/add_deps_test.go
diff --git a/appengine/cmd/dm/mutate/add_deps_test.go b/appengine/cmd/dm/mutate/add_deps_test.go
index 6a5628260e78d94cb2b726e45aac80b79500073e..0aa5113a176f9571825dd4a497640f4860335713 100644
--- a/appengine/cmd/dm/mutate/add_deps_test.go
+++ b/appengine/cmd/dm/mutate/add_deps_test.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/luci/gae/impl/memory"
 	"github.com/luci/gae/service/datastore"
+	"github.com/luci/luci-go/appengine/cmd/dm/distributor"
 	"github.com/luci/luci-go/appengine/cmd/dm/model"
 	"github.com/luci/luci-go/common/api/dm/service/v1"
 	. "github.com/luci/luci-go/common/testing/assertions"
@@ -84,14 +85,22 @@ func TestAddDeps(t *testing.T) {
 
 					So(ds.Get(a), ShouldBeNil)
 					So(ds.GetMulti(fds), ShouldBeNil)
-					So(a.AddingDepsBitmap.Size(), ShouldEqual, len(fds))
-					So(a.WaitingDepBitmap.Size(), ShouldEqual, len(fds))
-					So(a.State, ShouldEqual, dm.Attempt_ADDING_DEPS)
+					So(a.DepMap.Size(), ShouldEqual, len(fds))
+					So(a.State, ShouldEqual, dm.Attempt_EXECUTING)
 					So(fds[0].ForExecution, ShouldEqual, 1)
+
+					muts, err = (&FinishExecution{
+						ad.Auth.Id, &distributor.TaskResult{PersistentState: "hi"},
+					}).RollForward(c)
+					So(err, ShouldBeNil)
+					So(muts, ShouldBeNil)
+
+					So(ds.Get(a), ShouldBeNil)
+					So(a.State, ShouldEqual, dm.Attempt_WAITING)
 				})
 
 				Convey("adding new Attempts at the same time", func() {
-					ad.Atmpts = dm.NewAttemptList(map[string][]uint32{
+					ad.Attempts = dm.NewAttemptList(map[string][]uint32{
 						"to": {2, 3},
 						"tp": {1},
 					})
@@ -106,10 +115,18 @@ func TestAddDeps(t *testing.T) {
 
 					So(ds.Get(a), ShouldBeNil)
 					So(ds.GetMulti(fds), ShouldBeNil)
-					So(a.AddingDepsBitmap.Size(), ShouldEqual, len(fds))
-					So(a.WaitingDepBitmap.Size(), ShouldEqual, len(fds))
-					So(a.State, ShouldEqual, dm.Attempt_ADDING_DEPS)
+					So(a.DepMap.Size(), ShouldEqual, len(fds))
+					So(a.State, ShouldEqual, dm.Attempt_EXECUTING)
 					So(fds[0].ForExecution, ShouldEqual, 1)
+
+					muts, err = (&FinishExecution{
+						ad.Auth.Id, &distributor.TaskResult{PersistentState: "hi"},
+					}).RollForward(c)
+					So(err, ShouldBeNil)
+					So(muts, ShouldBeNil)
+
+					So(ds.Get(a), ShouldBeNil)
+					So(a.State, ShouldEqual, dm.Attempt_WAITING)
 				})
 			})
 		})