client: check if prev alloc is already terminated

This is a defensive fast-path as 7c6aa0be already fixed the deadlock.

client/allocwatcher/alloc_watcher.go

@@ -190,7 +190,12 @@ func (p *localPrevAlloc) Wait(ctx context.Context) error {
 
 	defer p.prevListener.Close()
 
-	//TODO Check p.aloc.Terminated() here
+	// Don't bother blocking for updates from the previous alloc if it has
+	// already terminated.
+	if p.prevStatus.Terminated() {
+		p.logger.Trace("previous allocation already terminated")
+		return nil
+	}
 
 	// Block until previous alloc exits
 	p.logger.Debug("waiting for previous alloc to terminate")

client/allocwatcher/alloc_watcher_test.go

@@ -71,6 +71,7 @@ func newConfig(t *testing.T) (Config, func()) {
 	alloc.PreviousAllocation = prevAR.Alloc().ID
 	alloc.Job.TaskGroups[0].EphemeralDisk.Sticky = true
 	alloc.Job.TaskGroups[0].EphemeralDisk.Migrate = true
+	alloc.Job.TaskGroups[0].Tasks[0].Driver = "mock_driver"
 
 	config := Config{
 		Alloc:          alloc,
@@ -114,15 +115,14 @@ func TestPrevAlloc_Noop(t *testing.T) {
 	<-done
 }
 
-// TestPrevAlloc_LocalPrevAlloc asserts that when a previous alloc runner is
-// set a localPrevAlloc will block on it.
-func TestPrevAlloc_LocalPrevAlloc(t *testing.T) {
+// TestPrevAlloc_LocalPrevAlloc_Block asserts that when a previous alloc runner
+// is set a localPrevAlloc will block on it.
+func TestPrevAlloc_LocalPrevAlloc_Block(t *testing.T) {
 	t.Parallel()
 	conf, cleanup := newConfig(t)
 
 	defer cleanup()
 
-	conf.Alloc.Job.TaskGroups[0].Tasks[0].Driver = "mock_driver"
 	conf.Alloc.Job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
 		"run_for": "500ms",
 	}
@@ -182,6 +182,25 @@ func TestPrevAlloc_LocalPrevAlloc(t *testing.T) {
 	})
 }
 
+// TestPrevAlloc_LocalPrevAlloc_Terminated asserts that when a previous alloc
+// runner has already terminated the watcher does not block on the broadcaster.
+func TestPrevAlloc_LocalPrevAlloc_Terminated(t *testing.T) {
+	t.Parallel()
+	conf, cleanup := newConfig(t)
+	defer cleanup()
+
+	conf.PreviousRunner.Alloc().ClientStatus = structs.AllocClientStatusComplete
+
+	waiter := NewAllocWatcher(conf)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	// Since prev alloc is terminal, Wait should exit immediately with no
+	// context error
+	require.NoError(t, waiter.Wait(ctx))
+}
+
 // TestPrevAlloc_StreamAllocDir_Ok asserts that streaming a tar to an alloc dir
 // works.
 func TestPrevAlloc_StreamAllocDir_Ok(t *testing.T) {