✨ Add the possibility to set a timeout for waiting job completion (#76)

<!--
Copyright (C) 2020-2022 Arm Limited or its affiliates and Contributors.
All rights reserved.
SPDX-License-Identifier: Proprietary
-->
### Description

- add a timeout entry and determine retries based on this value


### Test Coverage

<!--
Please put an `x` in the correct box e.g. `[x]` to indicate the testing
coverage of this change.
-->

- [x]  This change is covered by existing or additional automated tests.
- [ ] Manual testing has been performed (and evidence provided) as
automated testing was not feasible.
- [ ] Additional tests are not required for this change (e.g.
documentation update).

changes/20240619155022.feature

@@ -0,0 +1 @@
+:sparkles: Add the possibility to set a timeout for waiting job completion

utils/job/interfaces.go

@@ -8,6 +8,7 @@ package job
 
 import (
  "context"
+ "time"
 
  "github.com/ARM-software/embedded-development-services-client-utils/utils/resource"
 )
@@ -43,6 +44,8 @@ type IJobManager interface {
  HasJobCompleted(ctx context.Context, job IAsynchronousJob) (completed bool, err error)
  // HasJobStarted calls the services to determine whether the job has started.
  HasJobStarted(ctx context.Context, job IAsynchronousJob) (completed bool, err error)
- // WaitForJobCompletion waits for a job to complete.
+ // WaitForJobCompletion waits for a job to complete. Similar to WaitForJobCompletionWithTimeout but with a timeout set to 5 minutes.
  WaitForJobCompletion(ctx context.Context, job IAsynchronousJob) (err error)
+ // WaitForJobCompletionWithTimeout waits for a job to complete but with timeout protection.
+ WaitForJobCompletionWithTimeout(ctx context.Context, job IAsynchronousJob, timeout time.Duration) (err error)
 }

utils/job/manager.go

@@ -8,6 +8,7 @@ package job
 import (
  "context"
  "fmt"
+ "math"
  "net/http"
  "time"
 
@@ -55,19 +56,24 @@ func (m *Manager) FetchJobMessagesFirstPage(ctx context.Context, job IAsynchrono
  return
 }
 
-func waitForJobState(ctx context.Context, logger messages.IMessageLogger, job IAsynchronousJob, jobState string, checkStateFunc func(context.Context, IAsynchronousJob) (bool, error)) (err error) {
+func waitForJobState(ctx context.Context, logger messages.IMessageLogger, job IAsynchronousJob, jobState string, checkStateFunc func(context.Context, IAsynchronousJob) (bool, error), timeout time.Duration) (err error) {
  err = parallelisation.DetermineContextError(ctx)
  if err != nil {
  return
  }
 
+ subCtx, cancel := context.WithTimeout(ctx, timeout)
+ defer cancel()
+ retryCfg := retry.DefaultExponentialBackoffRetryPolicyConfiguration()
+ retryCfg.RetryMax = int(float64(timeout.Milliseconds())/math.Max(float64(retryCfg.RetryWaitMin.Milliseconds()), 1)) + 1
+
  jobName, err := job.FetchName()
  if err != nil {
  return
  }
  notStartedError := fmt.Errorf("%w: job [%v] has not reached the expected state [%v]", commonerrors.ErrCondition, jobName, jobState)
- err = retry.RetryOnError(ctx, logs.NewPlainLogrLoggerFromLoggers(logger), retry.DefaultExponentialBackoffRetryPolicyConfiguration(), func() error {
- inState, subErr := checkStateFunc(ctx, job)
+ err = retry.RetryOnError(subCtx, logs.NewPlainLogrLoggerFromLoggers(logger), retryCfg, func() error {
+ inState, subErr := checkStateFunc(subCtx, job)
  if subErr != nil {
  return subErr
  }
@@ -79,12 +85,12 @@ func waitForJobState(ctx context.Context, logger messages.IMessageLogger, job IA
  return
 }
 
-func (m *Manager) waitForJobToStart(ctx context.Context, logger messages.IMessageLogger, job IAsynchronousJob) error {
- return waitForJobState(ctx, logger, job, "start", m.HasJobStarted)
+func (m *Manager) waitForJobToStart(ctx context.Context, logger messages.IMessageLogger, job IAsynchronousJob, timeout time.Duration) error {
+ return waitForJobState(ctx, logger, job, "start", m.HasJobStarted, timeout)
 }
 
-func (m *Manager) waitForJobToHaveMessagesAvailable(ctx context.Context, logger messages.IMessageLogger, job IAsynchronousJob) error {
- return waitForJobState(ctx, logger, job, "have messages", m.areThereMessages)
+func (m *Manager) waitForJobToHaveMessagesAvailable(ctx context.Context, logger messages.IMessageLogger, job IAsynchronousJob, timeout time.Duration) error {
+ return waitForJobState(ctx, logger, job, "have messages", m.areThereMessages, timeout)
 }
 
 func (m *Manager) createMessagePaginator(ctx context.Context, job IAsynchronousJob) (paginator pagination.IStreamPaginatorAndPageFetcher, err error) {
@@ -94,7 +100,11 @@ func (m *Manager) createMessagePaginator(ctx context.Context, job IAsynchronousJ
  return
 }
 
-func (m *Manager) WaitForJobCompletion(ctx context.Context, job IAsynchronousJob) (err error) {
+func (m *Manager) WaitForJobCompletion(ctx context.Context, job IAsynchronousJob) error {
+ return m.WaitForJobCompletionWithTimeout(ctx, job, 5*time.Minute)
+}
+
+func (m *Manager) WaitForJobCompletionWithTimeout(ctx context.Context, job IAsynchronousJob, timeout time.Duration) (err error) {
  err = parallelisation.DetermineContextError(ctx)
  if err != nil {
  return
@@ -108,15 +118,17 @@ func (m *Manager) WaitForJobCompletion(ctx context.Context, job IAsynchronousJob
  _ = messageLogger.Close()
  }
  }()
- err = m.waitForJobToStart(ctx, messageLogger, job)
+ subCtx, cancel := context.WithTimeout(ctx, timeout)
+ defer cancel()
+ err = m.waitForJobToStart(subCtx, messageLogger, job, timeout)
  if err != nil {
  return
  }
- err = m.waitForJobToHaveMessagesAvailable(ctx, messageLogger, job)
+ err = m.waitForJobToHaveMessagesAvailable(subCtx, messageLogger, job, timeout)
  if err != nil {
  return
  }
- messagePaginator, err := m.createMessagePaginator(ctx, job)
+ messagePaginator, err := m.createMessagePaginator(subCtx, job)
  if err != nil {
  return
  }
@@ -126,7 +138,7 @@ func (m *Manager) WaitForJobCompletion(ctx context.Context, job IAsynchronousJob
  }
  }()
 
- wait, gCtx := errgroup.WithContext(ctx)
+ wait, gCtx := errgroup.WithContext(subCtx)
  wait.Go(func() error {
  return messageLogger.LogMessagesCollection(gCtx, messagePaginator)
 
@@ -138,7 +150,7 @@ func (m *Manager) WaitForJobCompletion(ctx context.Context, job IAsynchronousJob
  if err != nil {
  messageLogger.LogError(err)
  }
- _, err = m.HasJobCompleted(ctx, job)
+ _, err = m.HasJobCompleted(subCtx, job)
  return
 }
 

utils/job/manager_test.go

@@ -24,6 +24,8 @@ import (
  pagination2 "github.com/ARM-software/embedded-development-services-client-utils/utils/pagination"
  "github.com/ARM-software/golang-utils/utils/collection/pagination"
  "github.com/ARM-software/golang-utils/utils/commonerrors"
+ "github.com/ARM-software/golang-utils/utils/commonerrors/errortest"
+ "github.com/ARM-software/golang-utils/utils/field"
 )
 
 func TestManager_HasJobCompleted(t *testing.T) {
@@ -77,7 +79,7 @@ func TestManager_HasJobCompleted(t *testing.T) {
  assert.Equal(t, test.expectCompleted, completed)
  } else {
  assert.Error(t, err)
- assert.True(t, commonerrors.Any(err, test.expectedError))
+ errortest.AssertError(t, err, test.expectedError)
  assert.Equal(t, test.expectCompleted, completed)
  }
  })
@@ -132,7 +134,7 @@ func TestManager_checkForMessageStreamExhaustion(t *testing.T) {
  assert.True(t, messagePaginator.IsRunningDry())
  } else {
  assert.Error(t, err)
- assert.True(t, commonerrors.Any(err, test.expectedError))
+ errortest.AssertError(t, err, test.expectedError)
  }
  })
  }
@@ -149,15 +151,17 @@ func TestManager_WaitForJobCompletion(t *testing.T) {
  defer goleak.VerifyNone(t)
  tests := []struct {
  jobFunc func() (IAsynchronousJob, error)
- expectedError error
+ expectedError []error
+ timeout *time.Duration
  }{
  {
  jobFunc: mapFunc(jobtest.NewMockFailedAsynchronousJob),
- expectedError: commonerrors.ErrInvalid,
+ expectedError: []error{commonerrors.ErrInvalid},
  },
  {
  jobFunc: mapFunc(jobtest.NewMockQueuedAsynchronousJob),
- expectedError: commonerrors.ErrCondition,
+ expectedError: []error{commonerrors.ErrCondition, commonerrors.ErrTimeout, commonerrors.ErrCancelled},
+ timeout: field.ToOptionalDuration(500 * time.Millisecond),
  },
  {
  jobFunc: mapFunc(jobtest.NewMockSuccessfulAsynchronousJob),
@@ -178,17 +182,58 @@ func TestManager_WaitForJobCompletion(t *testing.T) {
 
  require.NoError(t, err)
  require.NotNil(t, factory)
- err = factory.WaitForJobCompletion(context.TODO(), job)
+ if test.timeout == nil {
+ err = factory.WaitForJobCompletion(context.TODO(), job)
+ } else {
+ err = factory.WaitForJobCompletionWithTimeout(context.TODO(), job, *test.timeout)
+ }
  if test.expectedError == nil {
  assert.NoError(t, err)
  } else {
  assert.Error(t, err)
- assert.True(t, commonerrors.Any(err, test.expectedError))
+ errortest.AssertError(t, err, test.expectedError...)
  }
  })
  }
 }
 
+func TestManager_WaitForJobCompletionTimeout(t *testing.T) {
+ defer goleak.VerifyNone(t)
+ tests := []struct {
+ jobFunc func() (IAsynchronousJob, error)
+ }{
+ {
+ jobFunc: mapFunc(jobtest.NewMockFailedAsynchronousJob),
+ },
+ {
+ jobFunc: mapFunc(jobtest.NewMockQueuedAsynchronousJob),
+ },
+ {
+ jobFunc: mapFunc(jobtest.NewMockSuccessfulAsynchronousJob),
+ },
+ }
+ for i := range tests {
+ test := tests[i]
+
+ t.Run(fmt.Sprintf("#%v", i), func(t *testing.T) {
+ // t.Parallel()
+ logger, err := logging.NewStandardClientLogger(fmt.Sprintf("test #%v", i), nil)
+ require.NoError(t, err)
+ loggerF := messages.NewMessageLoggerFactory(logger, false, time.Nanosecond)
+ job, err := test.jobFunc()
+ runOut := time.Nanosecond
+ factory, err := newMockJobManager(loggerF, time.Nanosecond, &runOut, job, err)
+
+ require.NoError(t, err)
+ require.NotNil(t, factory)
+
+ err = factory.WaitForJobCompletionWithTimeout(context.TODO(), job, time.Nanosecond)
+ assert.Error(t, err)
+ errortest.AssertError(t, err, commonerrors.ErrTimeout, commonerrors.ErrCancelled, commonerrors.ErrCondition)
+ })
+ }
+}
+
 func newMockJobManager(logger *messages.MessageLoggerFactory, backOffPeriod time.Duration, messagePaginatorRunOutTimeout *time.Duration, job IAsynchronousJob, errToReturn error) (*Manager, error) {
  pageNumber := rand.Intn(50) //nolint:gosec //causes G404: Use of weak random number generator
  messageStream := messages.NewMockMessagePaginatorFactory(pageNumber)