From f483906b9882c309b3cc1d3c81f6982778384b05 Mon Sep 17 00:00:00 2001
From: Laurent Senta <laurent@singulargarden.com>
Date: Thu, 28 Jul 2022 17:30:03 +0200
Subject: [PATCH] runner/docker: introduce collect timeout

---
 pkg/runner/local_docker.go | 29 ++++++++++++++++++++++++-----
 1 file changed, 24 insertions(+), 5 deletions(-)

diff --git a/pkg/runner/local_docker.go b/pkg/runner/local_docker.go
index 17dd605de..2403480f4 100644
--- a/pkg/runner/local_docker.go
+++ b/pkg/runner/local_docker.go
@@ -71,6 +71,9 @@ type LocalDockerRunnerConfig struct {
 	Ulimits []string `toml:"ulimits"`
 
 	ExposedPorts ExposedPorts `toml:"exposed_ports"`
+	// Collection timeout is the time we wait for the sync service to send us the test outcomes after
+	// all instances have finished.
+	OutcomesCollectionTimeout time.Duration `toml:"outcomes_collection_timeout"`
 }
 
 type testContainerInstance struct {
@@ -82,10 +85,11 @@ type testContainerInstance struct {
 // defaultConfig is the default configuration. Incoming configurations will be
 // merged with this object.
 var defaultConfig = LocalDockerRunnerConfig{
-	KeepContainers: false,
-	Unstarted:      false,
-	Background:     false,
-	Ulimits:        []string{"nofile=1048576:1048576"},
+	KeepContainers:            false,
+	Unstarted:                 false,
+	Background:                false,
+	Ulimits:                   []string{"nofile=1048576:1048576"},
+	OutcomesCollectionTimeout: time.Minute * 2,
 }
 
 // LocalDockerRunner is a runner that manually stands up as many docker
@@ -480,9 +484,14 @@ func (r *LocalDockerRunner) Run(ctx context.Context, input *api.RunInput, ow *rp
 	}
 
 	// ## Start the containers & log their outputs.
+	collectCtx, cancelCollect := context.WithCancel(ctx)
+
+	defer func() {
+		cancelCollect()
+	}()
 
 	// First we collect every container outcomes.
-	outcomesCollectIsCompleteCh, err := r.collectOutcomes(ctx, result, &template)
+	outcomesCollectIsCompleteCh, err := r.collectOutcomes(collectCtx, result, &template)
 	if err != nil {
 		log.Error(err)
 		return
@@ -633,6 +642,7 @@ func (r *LocalDockerRunner) Run(ctx context.Context, input *api.RunInput, ow *rp
 	// - we reach a timeout.
 
 	containersAreCompleteCh := make(chan bool)
+	outcomesCollectTimeout := make(chan bool)
 
 	// Wait for the containers
 	go func() {
@@ -640,6 +650,11 @@ func (r *LocalDockerRunner) Run(ctx context.Context, input *api.RunInput, ow *rp
 		containersAreCompleteCh <- true
 	}()
 
+	startOutcomesCollectTimeout := func() {
+		time.Sleep(cfg.OutcomesCollectionTimeout)
+		outcomesCollectTimeout <- true
+	}
+
 	waitingForContainers := true
 	waitingForOutcomes := true
 
@@ -649,9 +664,13 @@ func (r *LocalDockerRunner) Run(ctx context.Context, input *api.RunInput, ow *rp
 		case <-containersAreCompleteCh:
 			log.Infow("all containers are complete")
 			waitingForContainers = false
+			go startOutcomesCollectTimeout()
 		case <-outcomesCollectIsCompleteCh:
 			log.Infow("all outcomes are complete")
 			waitingForOutcomes = false
+		case <-outcomesCollectTimeout:
+			log.Infow("we timeout'd waiting for outcomes")
+			waitingForOutcomes = false
 		case <-ctx.Done():
 			log.Infow("test run is canceled")
 			return