fix run loop

orchestrator/scheduler/scheduler.go

@@ -99,6 +99,9 @@ func (s *Scheduler) Update(msg loop.Msg) loop.Cmd {
 			)
 		}
 
+		cmds = append(cmds,
+			work.CmdScheduleNextJob(),
+		)
 		if s.ExecOutWalker != nil {
 			cmds = append(cmds, execout.CmdDownloadSegment(0))
 		}
@@ -133,12 +136,14 @@ func (s *Scheduler) Update(msg loop.Msg) loop.Cmd {
 	case stage.MsgMergeFinished:
 		s.Stages.MergeCompleted(msg.Unit)
 		cmds = append(cmds,
+			work.CmdScheduleNextJob(),
 			s.Stages.CmdTryMerge(msg.Stage),
 		)
 
 	case stage.MsgAllStoresCompleted:
 		s.storesSyncCompleted = true
 		cmds = append(cmds,
+			work.CmdScheduleNextJob(), // in case some mapper jobs need scheduling
 			s.cmdShutdownWhenComplete(),
 		)
 

orchestrator/stage/stages.go

@@ -162,14 +162,12 @@ func (s *Stages) AllStoresCompleted() bool {
 	}
 	lastSegment := s.storeSegmenter.LastIndex()
 
-	fmt.Println("----------------------------------")
 	for idx, stage := range s.stages {
 		if stage.kind != KindStore {
 			continue
 		}
 		for seg := s.storeSegmenter.FirstIndex(); seg <= lastSegment; seg++ {
 			state := s.getState(Unit{Segment: seg, Stage: idx})
-			fmt.Println("stage: ", idx, "segment: ", seg, "status:", state)
 			if state != UnitCompleted && state != UnitNoOp {
 				return false
 			}

test/runnable_test.go

@@ -147,7 +147,6 @@ func (f *testRun) run(t *testing.T, testName string) error {
 	count := 0
 	workerFactory := func(ctx context.Context) work.Worker {
 		count++
-		fmt.Println("factoring worker ", count)
 		return &TestWorker{
 			t:                      t,
 			responseCollector:      newResponseCollector(ctx),