Workers should fail to start if they were stopped before (#450)

internal/internal_worker.go

@@ -890,6 +890,7 @@ func (aw *AggregatedWorker) RegisterActivityWithOptions(a interface{}, options R
 
 // Start the worker in a non-blocking fashion.
 func (aw *AggregatedWorker) Start() error {
+	aw.assertNotStopped()
 	if err := initBinaryChecksum(); err != nil {
 		return fmt.Errorf("failed to get executable checksum: %v", err)
 	}
@@ -934,6 +935,18 @@ func (aw *AggregatedWorker) Start() error {
 	return nil
 }
 
+func (aw *AggregatedWorker) assertNotStopped() {
+	stopped := true
+	select {
+	case <-aw.stopC:
+	default:
+		stopped = false
+	}
+	if stopped {
+		panic("attempted to start a worker that has been stopped before")
+	}
+}
+
 var binaryChecksum string
 var binaryChecksumLock sync.Mutex
 

internal/internal_worker_test.go

@@ -1556,6 +1556,26 @@ func (s *internalWorkerTestSuite) TestWorkerStartFailsWithInvalidNamespace() {
 	}
 }
 
+func (s *internalWorkerTestSuite) TestStartWorkerAfterStopped() {
+	defer func() {
+		if r := recover(); r == nil {
+			assert.Fail(s.T(), "calling start after stop must result in panic")
+		}
+	}()
+	worker := createWorkerWithThrottle(s.service, 500.0, nil)
+	worker.RegisterActivity(testActivityNoResult)
+	worker.RegisterWorkflow(testWorkflowReturnStruct)
+	err := worker.Start()
+	require.NoError(s.T(), err)
+	time.Sleep(time.Millisecond * 200)
+	assert.True(s.T(), worker.activityWorker.worker.isWorkerStarted)
+	assert.True(s.T(), worker.workflowWorker.worker.isWorkerStarted)
+	worker.Stop()
+	assert.False(s.T(), worker.activityWorker.worker.isWorkerStarted)
+	assert.False(s.T(), worker.workflowWorker.worker.isWorkerStarted)
+	_ = worker.Start() // must panic
+}
+
 func ofPollActivityTaskQueueRequest(tps float64) gomock.Matcher {
 	return &mockPollActivityTaskQueueRequest{tps: tps}
 }