Initial distributor implementation

common/api/dm/service/v1/attempt_state_evolve.go

Index: common/api/dm/service/v1/attempt_state_evolve.go
diff --git a/common/api/dm/service/v1/attempt_state_evolve.go b/common/api/dm/service/v1/attempt_state_evolve.go
index 3a7642c7926bafdbb30f356e9d080b513ee071e9..451347be297611b731b7c1ff6444faad4cad3689 100644
--- a/common/api/dm/service/v1/attempt_state_evolve.go
+++ b/common/api/dm/service/v1/attempt_state_evolve.go
@@ -12,12 +12,23 @@ import (
 // transitions. The identity transition (X -> X) is implied, as long as X has an
 // entry in this mapping.
 var validAttemptStateEvolution = map[Attempt_State][]Attempt_State{
-	Attempt_ADDING_DEPS:              {Attempt_BLOCKED, Attempt_NEEDS_EXECUTION},
-	Attempt_BLOCKED:                  {Attempt_AWAITING_EXECUTION_STATE, Attempt_NEEDS_EXECUTION},
-	Attempt_AWAITING_EXECUTION_STATE: {Attempt_NEEDS_EXECUTION},
-	Attempt_EXECUTING:                {Attempt_ADDING_DEPS, Attempt_FINISHED},
-	Attempt_FINISHED:                 {},
-	Attempt_NEEDS_EXECUTION:          {Attempt_EXECUTING},
+	Attempt_SCHEDULING: {
+		Attempt_EXECUTING,         // scheduled
+		Attempt_ABNORMAL_FINISHED, // cancelled/timeout/err/etc.
+	},
+	Attempt_EXECUTING: {
+		Attempt_SCHEDULING,        // Retry
+		Attempt_WAITING,           // EnsureGraphData
+		Attempt_FINISHED,          // FinishAttempt
+		Attempt_ABNORMAL_FINISHED, // cancel/timeout/err/etc.
+	},
+	Attempt_WAITING: {
+		Attempt_SCHEDULING,        // unblocked
+		Attempt_ABNORMAL_FINISHED, // cancelled
+	},
+
+	Attempt_FINISHED:          {},
+	Attempt_ABNORMAL_FINISHED: {},
 }
 
 // Evolve attempts to evolve the state of this Attempt. If the state evolution