Repository: flant/shell-operator
Pull request #406 (2022-06-13 17:02:44)

chore(deps): bump actions/cache from 3.0.2 to 3.0.4
Bumps [actions/cache](https://github.com/actions/cache) from 3.0.2 to 3.0.4.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/actions/cache/releases">actions/cache's releases</a>.</em></p>
<blockquote>
<h2>v3.0.4</h2>
<p>In this release, we have fixed the tar creation error while trying to create it with path as <code>~/</code> home folder on <code>ubuntu-latest</code>.</p>
<h2>v3.0.3</h2>
<p>Fixed avoiding empty cache save when no files are available for caching. (<a href="https://github-redirect.dependabot.com/actions/cache/issues/624">actions/cache#624</a>)</p>
</blockquote>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/actions/cache/blob/main/RELEASES.md">actions/cache's changelog</a>.</em></p>
<blockquote>
<h3>3.0.2</h3>
<ul>
<li>Added support for dynamic cache size cap on GHES.</li>
</ul>
<h3>3.0.3</h3>
<ul>
<li>Fixed avoiding empty cache save when no files are available for caching. (<a href="https://github-redirect.dependabot.com/actions/cache/issues/624">issue</a>)</li>
</ul>
<h3>3.0.4</h3>
<ul>
<li>Fixed tar creation error while trying to create tar with path as <code>~/</code> home folder on <code>ubuntu-latest</code>. (<a href="https://github-redirect.dependabot.com/actions/cache/issues/689">issue</a>)</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/actions/cache/commit/c3f1317a9e7b1ef106c153ac8c0f00fed3ddbc0d"><code>c3f1317</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/actions/cache/issues/813">#813</a> from actions/users/kotewar/upgrading-cache-to-v2.0.6</li>
<li><a href="https://github.com/actions/cache/commit/d0a54b996cb35d7beba3674d28753ff7ccf2430c"><code>d0a54b9</code></a> Fixed typo</li>
<li><a href="https://github.com/actions/cache/commit/8c5bd0c82db6e9fd53fc132c68d5f7649d568271"><code>8c5bd0c</code></a> Updated README file with release info</li>
<li><a href="https://github.com/actions/cache/commit/c9c0f735586a8329cd6973dfeb9ceaf13e033631"><code>c9c0f73</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/actions/cache/issues/812">#812</a> from actions/users/kotewar/upgrading-cache-to-v2.0.6</li>
<li><a href="https://github.com/actions/cache/commit/2b6caae3c9d06fc8e6b142308e16b3c6eaaf4637"><code>2b6caae</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/actions/cache/issues/495">#495</a> from ostera/patch-1</li>
<li><a href="https://github.com/actions/cache/commit/dd58d1315fbfa379835d53b0bc75d1a3420fcaf0"><code>dd58d13</code></a> Added release info and upgraded version</li>
<li><a href="https://github.com/actions/cache/commit/acace7fa1bd9ed59015a76e9de8a00ac88229033"><code>acace7f</code></a> Merge branch 'main' into patch-1</li>
<li><a href="https://github.com/actions/cache/commit/438628ac22a3596a70933e8dd4f06fe0c1550315"><code>438628a</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/actions/cache/issues/554">#554</a> from albertstill/improve-restore-key-docs</li>
<li><a href="https://github.com/actions/cache/commit/c296e6a08c6a639ec6e35fbfdcde61c3f6107885"><code>c296e6a</code></a> Updated <code>@​actions/cache</code> version in license file</li>
<li><a href="https://github.com/actions/cache/commit/7ed7f22ed87596cd0d1c13baca967dc49bd9dc6f"><code>7ed7f22</code></a> Updated actions/cache to v2.0.6</li>
<li>Additional commits viewable in <a href="https://github.com/actions/cache/compare/v3.0.2...v3.0.4">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=actions/cache&package-manager=github_actions&previous-version=3.0.2&new-version=3.0.4)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
index 958c595..f5a6ef4 100644
--- a/.github/workflows/build.yaml
+++ b/.github/workflows/build.yaml
@@ -18,7 +18,7 @@ jobs:
 
       - name: Restore Go modules
         id: go-modules-cache
-        uses: actions/cache@v3.0.2
+        uses: actions/cache@v3.0.4
         with:
           path: |
             ~/go/pkg/mod
diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml
index e1cec4e..7a53778 100644
--- a/.github/workflows/lint.yaml
+++ b/.github/workflows/lint.yaml
@@ -20,7 +20,7 @@ jobs:
 
       - name: Restore Go modules
         id: go-modules-cache
-        uses: actions/cache@v3.0.2
+        uses: actions/cache@v3.0.4
         with:
           path: |
             ~/go/pkg/mod
diff --git a/pkg/task/queue/task_queue.go b/pkg/task/queue/task_queue.go
index 1f15b5d..d555976 100644
--- a/pkg/task/queue/task_queue.go
+++ b/pkg/task/queue/task_queue.go
@@ -28,10 +28,10 @@ config parameter.
 */
 
 var (
-	WaitLoopCheckInterval = 125 * time.Millisecond
-	DelayOnQueueIsEmpty   = 250 * time.Millisecond
-	DelayOnFailedTask     = 5 * time.Second
-	DelayOnRepeat         = 25 * time.Millisecond
+	DefaultWaitLoopCheckInterval    = 125 * time.Millisecond
+	DefaultDelayOnQueueIsEmpty      = 250 * time.Millisecond
+	DefaultInitialDelayOnFailedTask = 5 * time.Second
+	DefaultDelayOnRepeat            = 25 * time.Millisecond
 )
 
 type TaskStatus string
@@ -76,11 +76,24 @@ type TaskQueue struct {
 
 	measureActionFn     func()
 	measureActionFnOnce sync.Once
+
+	// Timing settings.
+	WaitLoopCheckInterval time.Duration
+	DelayOnQueueIsEmpty   time.Duration
+	DelayOnRepeat         time.Duration
+	ExponentialBackoffFn  func(failureCount int) time.Duration
 }
 
 func NewTasksQueue() *TaskQueue {
 	return &TaskQueue{
 		items: make([]task.Task, 0),
+		// Default timings
+		WaitLoopCheckInterval: DefaultWaitLoopCheckInterval,
+		DelayOnQueueIsEmpty:   DefaultDelayOnQueueIsEmpty,
+		DelayOnRepeat:         DefaultDelayOnRepeat,
+		ExponentialBackoffFn: func(failureCount int) time.Duration {
+			return exponential_backoff.CalculateDelay(DefaultInitialDelayOnFailedTask, failureCount)
+		},
 	}
 }
 
@@ -404,7 +417,7 @@ func (q *TaskQueue) Start() {
 			switch taskRes.Status {
 			case Fail:
 				// Exponential backoff delay before retry.
-				nextSleepDelay = exponential_backoff.CalculateDelay(DelayOnFailedTask, t.GetFailureCount())
+				nextSleepDelay = q.ExponentialBackoffFn(t.GetFailureCount())
 				t.IncrementFailureCount()
 				q.Status = fmt.Sprintf("sleep after fail for %s", nextSleepDelay.String())
 			case Success, Keep:
@@ -431,7 +444,7 @@ func (q *TaskQueue) Start() {
 				q.Status = ""
 			case Repeat:
 				// repeat a current task after a small delay
-				nextSleepDelay = DelayOnRepeat
+				nextSleepDelay = q.DelayOnRepeat
 				q.Status = "repeat head task"
 			}
 
@@ -470,30 +483,33 @@ func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task {
 
 	// Initialize wait settings.
 	waitBegin := time.Now()
-	waitUntil := DelayOnQueueIsEmpty
+	waitUntil := q.DelayOnQueueIsEmpty
 	if sleepDelay != 0 {
 		waitUntil = sleepDelay
 	}
 
-	checkTicker := time.NewTicker(WaitLoopCheckInterval)
+	checkTicker := time.NewTicker(q.WaitLoopCheckInterval)
 	q.waitMu.Lock()
 	q.waitInProgress = true
 	q.cancelDelay = false
 	q.waitMu.Unlock()
 
+	origStatus := q.Status
+
 	defer func() {
 		checkTicker.Stop()
 		q.waitMu.Lock()
 		q.waitInProgress = false
 		q.cancelDelay = false
 		q.waitMu.Unlock()
+		q.Status = origStatus
 	}()
 
 	// Wait for the queued task with some delay.
 	// Every tick increases the 'elapsed' counter until it outgrows the waitUntil value.
 	// Or, delay can be canceled to handle new head task immediately.
-	origStatus := q.Status
 	for {
+		checkTask := false
 		select {
 		case <-q.ctx.Done():
 			// Queue is stopped.
@@ -502,36 +518,42 @@ func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task {
 			// Check and update waitUntil.
 			elapsed := time.Since(waitBegin)
 
-			cancelDelay := false
 			q.waitMu.Lock()
 			if q.cancelDelay {
-				cancelDelay = true
+				// Reset waitUntil to check task immediately.
+				waitUntil = elapsed
 			}
 			q.waitMu.Unlock()
 
 			// Wait loop is done or canceled: break select to check for the head task.
-			if elapsed >= waitUntil || cancelDelay {
-				// Increase waitUntil to wait on the next iteration.
-				waitUntil += DelayOnQueueIsEmpty
-				break
+			if elapsed >= waitUntil {
+				// Increase waitUntil to wait on the next iteration and go check for the head task.
+				checkTask = true
 			}
+		}
 
-			// Wait loop still in progress: update queue status.
-			waitSeconds := time.Since(waitBegin).Truncate(time.Second).String()
-			if sleepDelay == 0 {
-				q.Status = fmt.Sprintf("waiting for task %s", waitSeconds)
+		// Break the for-loop to see if the head task can be returned.
+		if checkTask {
+			if q.IsEmpty() {
+				// No task to return: increase wait time.
+				waitUntil += q.DelayOnQueueIsEmpty
 			} else {
-				delay := sleepDelay.Truncate(time.Second).String()
-				q.Status = fmt.Sprintf("%s (%s left of %s delay)", origStatus, waitSeconds, delay)
+				return q.GetFirst()
 			}
 		}
-		if !q.IsEmpty() {
-			return q.GetFirst()
+
+		// Wait loop still in progress: update queue status.
+		waitSeconds := time.Since(waitBegin).Truncate(time.Second).String()
+		if sleepDelay == 0 {
+			q.Status = fmt.Sprintf("waiting for task %s", waitSeconds)
+		} else {
+			delay := sleepDelay.Truncate(time.Second).String()
+			q.Status = fmt.Sprintf("%s (%s left of %s delay)", origStatus, waitSeconds, delay)
 		}
 	}
 }
 
-// CancelTaskDelay breaks wait loop. Useful to break the possible maximum sleep delay.
+// CancelTaskDelay breaks wait loop. Useful to break the possible long sleep delay.
 func (q *TaskQueue) CancelTaskDelay() {
 	q.waitMu.Lock()
 	if q.waitInProgress {


Test patch:
diff --git a/.github/workflows/tests-labeled.yaml b/.github/workflows/tests-labeled.yaml
index f03881d..c5ca702 100644
--- a/.github/workflows/tests-labeled.yaml
+++ b/.github/workflows/tests-labeled.yaml
@@ -67,7 +67,7 @@ jobs:
 
       - name: Restore Go modules
         id: go-modules-cache
-        uses: actions/cache@v3.0.2
+        uses: actions/cache@v3.0.4
         with:
           path: |
             ~/go/pkg/mod
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 084bbbe..36d504c 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -19,7 +19,7 @@ jobs:
 
       - name: Restore Go modules
         id: go-modules-cache
-        uses: actions/cache@v3.0.2
+        uses: actions/cache@v3.0.4
         with:
           path: |
             ~/go/pkg/mod
diff --git a/pkg/task/queue/task_queue_test.go b/pkg/task/queue/task_queue_test.go
index 716f602..38bed0b 100644
--- a/pkg/task/queue/task_queue_test.go
+++ b/pkg/task/queue/task_queue_test.go
@@ -2,8 +2,10 @@ package queue
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	. "github.com/onsi/gomega"
 
@@ -72,3 +74,160 @@ func Test_TasksQueue_Remove(t *testing.T) {
 		ContainSubstring("task_03"),
 	))
 }
+
+func Test_ExponentialBackoff(t *testing.T) {
+	g := NewWithT(t)
+	// Init and prefill queue.
+	q := NewTasksQueue()
+	q.WithContext(context.TODO())
+	q.WithName("test-queue")
+	// Since we don't want the test to run for too long, we don't
+	// want to use lengthy times.
+	q.WaitLoopCheckInterval = 5 * time.Millisecond // default is 125ms
+	q.DelayOnQueueIsEmpty = 5 * time.Millisecond   // default is 250ms
+	q.DelayOnRepeat = 5 * time.Millisecond         // default is 25ms
+	// Add one task.
+	Task := &task.BaseTask{Id: "First one"}
+	q.AddFirst(Task)
+
+	// Set handler to fail 10 times and catch timestamps for each task execution.
+	runsAt := make([]time.Time, 0)
+	failureCounts := make([]int, 0)
+	const fails = 10
+	failsCount := fails
+	queueStopCh := make(chan struct{}, 1)
+	q.WithHandler(func(t task.Task) (res TaskResult) {
+		runsAt = append(runsAt, time.Now())
+		failureCounts = append(failureCounts, t.GetFailureCount())
+		if failsCount > 0 {
+			res.Status = Fail
+			failsCount--
+			return
+		}
+		res.Status = Success
+		res.AfterHandle = func() {
+			close(queueStopCh)
+		}
+		return
+	})
+
+	// Set exponential backoff to the constant delay just to wait more than DelayOnQueueIsEmpty.
+	// It is a test of delaying between task runs, not a test of exponential distribution.
+	mockExponentialDelay := 30 * time.Millisecond
+	q.ExponentialBackoffFn = func(failureCount int) time.Duration {
+		return mockExponentialDelay
+	}
+
+	q.Start()
+
+	// Expect TaskHandler returns Success result.
+	g.Eventually(queueStopCh, "5s", "20ms").Should(BeClosed(), "Should handle first task in queue successfully")
+
+	// Expect TaskHandler called 'fails' times.
+	g.Expect(Task.GetFailureCount()).Should(Equal(fails), "task should fail %d times.", fails)
+
+	prev := failureCounts[0]
+	for i := 1; i < len(failureCounts); i++ {
+		cur := failureCounts[i]
+		g.Expect(cur > prev).Should(BeTrue(), "TaskHandler should receive task with growing FailureCount. Got %d after %d", cur, prev)
+	}
+
+	// Expect mean delay is greater than mocked delay.
+	mean, _ := calculateMeanDelay(runsAt)
+	g.Expect(mean).Should(BeNumerically(">", mockExponentialDelay),
+		"mean delay of %d fails should be more than %s, got %s. Check exponential delaying not broken in Start or waitForTask.",
+		fails, mockExponentialDelay.String(), mean.Truncate(100*time.Microsecond).String())
+
+}
+
+func calculateMeanDelay(in []time.Time) (mean time.Duration, deltas []int64) {
+	var sum int64
+
+	// Calculate deltas from timestamps.
+	prev := in[0].UnixNano()
+	for i := 1; i < len(in); i++ {
+		delta := in[i].UnixNano() - prev
+		prev = in[i].UnixNano()
+		deltas = append(deltas, delta)
+		sum += delta
+	}
+	mean = time.Duration(sum / int64(len(deltas)))
+	return
+}
+
+func Test_CancelDelay(t *testing.T) {
+	g := NewWithT(t)
+	// Init and prefill queue.
+	q := NewTasksQueue()
+	q.WithContext(context.TODO())
+	q.WithName("test-queue")
+	// Since we don't want the test to run for too long, we don't
+	// want to use lengthy times.
+	q.WaitLoopCheckInterval = 5 * time.Millisecond // default is 125ms
+	q.DelayOnQueueIsEmpty = 5 * time.Millisecond   // default is 250ms
+	q.DelayOnRepeat = 5 * time.Millisecond         // default is 25ms
+	// Add 'always fail' task.
+	ErrTask := &task.BaseTask{Id: "erroneous"}
+	q.AddFirst(ErrTask)
+
+	HealingTask := &task.BaseTask{Id: "healing"}
+
+	// Set handler to always return fail for task "erroneous", and return success for other tasks.
+	// Catch time between "erroneous" task handling and "healing" task handling.
+	startedAt := time.Now()
+	endedAt := startedAt
+	delayStartsCh := make(chan struct{}, 1)
+	healingDoneCh := make(chan struct{}, 1)
+	q.WithHandler(func(t task.Task) (res TaskResult) {
+		if t.GetId() == ErrTask.GetId() {
+			res.Status = Fail
+			// Close chan after first delay.
+			if t.GetFailureCount() == 1 {
+				res.AfterHandle = func() {
+					close(delayStartsCh)
+				}
+			}
+			return
+		}
+		if t.GetId() == HealingTask.GetId() {
+			endedAt = time.Now()
+			res.AfterHandle = func() {
+				close(healingDoneCh)
+			}
+		}
+		res.Status = Success
+		return
+	})
+
+	// Set exponential backoff to the constant delay just to wait more than DelayOnQueueIsEmpty.
+	// It is a test of delaying between task runs, not a test of exponential distribution.
+	mockExponentialDelay := 150 * time.Millisecond
+	q.ExponentialBackoffFn = func(failureCount int) time.Duration {
+		return mockExponentialDelay
+	}
+
+	// Start handling 'erroneous' task.
+	q.Start()
+
+	// Expect TaskHandler returns Success result.
+	g.Eventually(delayStartsCh, "5s", "20ms").Should(BeClosed(), "Should handle failed task and starts a delay")
+
+	// Add healing task and cancel delay in parallel.
+	go func() {
+		q.AddFirst(HealingTask)
+		q.CancelTaskDelay()
+	}()
+
+	// Expect queue handles 'healing' task.
+	g.Eventually(healingDoneCh, "5s", "20ms").Should(BeClosed(), "Should handle 'healing' task")
+
+	elapsed := endedAt.Sub(startedAt).Truncate(100 * time.Microsecond)
+
+	// Expect elapsed is less than mocked delay.
+	g.Expect(elapsed).Should(BeNumerically(">", mockExponentialDelay),
+		"Should delay after failed task. Got delay of %s, expect more than %s. Check delay for failed task not broken in Start or waitForTask.",
+		elapsed.String(), mockExponentialDelay.String())
+	g.Expect(elapsed).Should(BeNumerically("<", 2*mockExponentialDelay),
+		"Should stop delaying after CancelTaskDelay call. Got delay of %s, expect less than %s. Check cancel delay not broken in Start or waitForTask.",
+		elapsed.String(), (2 * mockExponentialDelay).String())
+}