Do not update data source or task if task claim was lost

Author: darinkrauss

auth/service/service/client.go

@@ -88,8 +88,8 @@ func (c *Client) CreateProviderSession(ctx context.Context, create *auth.Provide
 
 	if err = prvdr.OnCreate(ctx, providerSession); err != nil {
 		log.LoggerFromContext(ctx).WithError(err).Error("Unable to finalize creation of provider session")
-		if err := c.deleteProviderSession(ctx, repository, providerSession); err != nil {
-			log.LoggerFromContext(ctx).WithError(err).Warn("Unable to delete provider session")
+		if deleteErr := c.deleteProviderSession(ctx, repository, providerSession); deleteErr != nil {
+			log.LoggerFromContext(ctx).WithError(deleteErr).Warn("Unable to delete provider session")
 		}
 		return nil, err
 	}

dexcom/fetch/runner.go

@@ -180,14 +180,17 @@ func (t *TaskRunner) Run(ctx context.Context) {
 		t.task.AppendError(err)
 	}
 
-	// A permanently failed task is not rescheduled, unless its outcome could not be recorded on the data source, in
-	// which case run again so a later run can record it
-	err := t.updateDataSourceWithTaskState()
-	if err != nil {
-		t.task.AppendError(err)
-	}
-	if err != nil || !t.task.IsFailed() {
-		t.task.RepeatAvailableAfter(pointer.Default(t.availableAfter, availableAfterDuration()))
+	// If we didn't lose the claim, then update data source and repeat if not failed
+	if !errors.Is(context.Cause(t.context), task.ErrClaimLost) {
+		err := t.updateDataSourceWithTaskState()
+		if err != nil {
+			t.task.AppendError(err)
+		}
+		if err != nil || !t.task.IsFailed() {
+			t.task.RepeatAvailableAfter(pointer.Default(t.availableAfter, availableAfterDuration()))
+		}
+	} else {
+		t.logger.Warn("Skipped updating data source and task because the task claim was lost")
 	}
 }
 

dexcom/fetch/runner_test.go

@@ -187,14 +187,16 @@ var _ = Describe("Runner", func() {
 
 		Context("with task runner and context", func() {
 			var taskRunner *dexcomFetch.TaskRunner
+			var logger *logTest.Logger
 			var ctx context.Context
 
 			BeforeEach(func() {
 				var err error
 				taskRunner, err = dexcomFetch.NewTaskRunner(provider, tsk)
 				Expect(err).ToNot(HaveOccurred())
 				Expect(taskRunner).ToNot(BeNil())
-				ctx = log.NewContextWithLogger(context.Background(), logTest.NewLogger())
+				logger = logTest.NewLogger()
+				ctx = log.NewContextWithLogger(context.Background(), logger)
 			})
 
 			assertTaskState := func(state string) {
@@ -364,6 +366,21 @@ var _ = Describe("Runner", func() {
 					assertTaskAndDataSourceError(dexcomFetch.ErrorCodeResourceFailure, "unable to get provider session")
 				})
 
+				It("discards the run outcome if the task claim is lost", func() {
+					claimContext, claimCancel := context.WithCancelCause(ctx)
+					defer claimCancel(nil)
+					testErr := errorsTest.RandomError()
+					authClient.EXPECT().GetProviderSession(matchContext(), "test-provider-session-id").DoAndReturn(func(ctx context.Context, id string) (*auth.ProviderSession, error) {
+						claimCancel(task.ErrClaimLost)
+						return nil, testErr
+					})
+					taskRunner.Run(claimContext)
+					assertTaskState(task.TaskStateRunning)
+					Expect(dataSrc.State).To(Equal(dataSource.StateConnected))
+					Expect(dataSrc.HasError()).To(BeFalse())
+					logger.AssertWarn("Skipped updating data source and task because the task claim was lost")
+				})
+
 				It("fails if the provider session is missing", func() {
 					authClient.EXPECT().GetProviderSession(matchContext(), "test-provider-session-id").Return(nil, nil)
 					dataSourceClient.EXPECT().Update(matchContext(), "test-data-source-id", matchNil(), matchNotNil()).DoAndReturn(mockDataSourceClientUpdate(dataSrc))
@@ -643,7 +660,78 @@ var _ = Describe("Runner", func() {
 					// deviceHashes - not in data
 					// dataSource.LatestDataTime - not nil (recent)
 					// refresh token
-					// data ranges multiple 30 day segments
+				})
+
+				Context("with provider session and a data range spanning multiple chunks", func() {
+					var providerSession *auth.ProviderSession
+					var firstChunkStartTime time.Time
+					var firstChunkEndTime time.Time
+					var secondChunkEndTime time.Time
+
+					BeforeEach(func() {
+						providerSession = &auth.ProviderSession{
+							ID:     "test-provider-session-id",
+							UserID: "test-user-id",
+							OAuthToken: &auth.OAuthToken{
+								AccessToken:    "test-access-token-1",
+								TokenType:      "Bearer",
+								RefreshToken:   "test-refresh-token-1",
+								ExpirationTime: time.Now().Add(time.Minute),
+							},
+						}
+						authClient.EXPECT().GetProviderSession(matchContext(), "test-provider-session-id").Return(providerSession, nil)
+						authClient.EXPECT().UpdateProviderSession(matchContext(), "test-provider-session-id", matchNotNil()).DoAndReturn(mockAuthClientUpdateProviderSession(providerSession)).AnyTimes()
+						firstChunkStartTime = time.Now().Add(-45 * Day)
+						firstChunkEndTime = firstChunkStartTime.AddDate(0, 0, dexcomFetch.DataRangeDaysMaximum)
+						secondChunkEndTime = time.Now().Add(-3 * Day)
+						dataRangeResponse := &dexcom.DataRangesResponse{
+							Calibrations: &dexcom.DataRange{
+								Start: &dexcom.Moment{SystemTime: &dexcom.Time{Time: firstChunkStartTime}},
+								End:   &dexcom.Moment{SystemTime: &dexcom.Time{Time: secondChunkEndTime}},
+							},
+						}
+						dexcomClient.EXPECT().GetDataRange(matchContext(), nil, matchNotNil()).DoAndReturn(mockDexcomClientGetDataRange(nil, dataRangeResponse, nil))
+					})
+
+					// Expects the fetch of a single chunk, all responses empty, invoking onEvents, if any, during the
+					// final fetch of the chunk
+					expectFetchChunk := func(startTime time.Time, endTime time.Time, onEvents func()) {
+						dexcomClient.EXPECT().GetAlerts(matchContext(), startTime, endTime, matchNotNil()).DoAndReturn(mockDexcomClientGetData(nil, &dexcom.AlertsResponse{Records: &dexcom.Alerts{}}, nil))
+						dexcomClient.EXPECT().GetCalibrations(matchContext(), startTime, endTime, matchNotNil()).DoAndReturn(mockDexcomClientGetData(nil, &dexcom.CalibrationsResponse{Records: &dexcom.Calibrations{}}, nil))
+						dexcomClient.EXPECT().GetDevices(matchContext(), startTime, endTime, matchNotNil()).DoAndReturn(mockDexcomClientGetData(nil, &dexcom.DevicesResponse{Records: &dexcom.Devices{}}, nil))
+						dexcomClient.EXPECT().GetEGVs(matchContext(), startTime, endTime, matchNotNil()).DoAndReturn(mockDexcomClientGetData(nil, &dexcom.EGVsResponse{Records: &dexcom.EGVs{}}, nil))
+						dexcomClient.EXPECT().GetEvents(matchContext(), startTime, endTime, matchNotNil()).DoAndReturn(func(ctx context.Context, startTime time.Time, endTime time.Time, tokenSource oauth.TokenSource) (*dexcom.EventsResponse, error) {
+							if onEvents != nil {
+								onEvents()
+							}
+							return &dexcom.EventsResponse{Records: &dexcom.Events{}}, nil
+						})
+					}
+
+					It("fetches every chunk of the data range", func() {
+						expectFetchChunk(firstChunkStartTime, firstChunkEndTime, nil)
+						expectFetchChunk(firstChunkEndTime, secondChunkEndTime, nil)
+						dataSourceClient.EXPECT().Update(matchContext(), "test-data-source-id", matchNil(), matchNotNil()).DoAndReturn(mockDataSourceClientUpdate(dataSrc))
+						taskRunner.Run(ctx)
+						assertTaskAndDataSourceState(task.TaskStatePending)
+						assertTaskAvailableAfterStandardDuration()
+						assertTaskRetryCountNotPresent()
+						assertTaskAndDataSourceErrorNotPresent()
+						assertDataSourceLastImportTimePresent()
+					})
+
+					It("discards the run outcome if the task claim is lost mid-fetch", func() {
+						claimContext, claimCancel := context.WithCancelCause(ctx)
+						defer claimCancel(nil)
+						expectFetchChunk(firstChunkStartTime, firstChunkEndTime, func() { claimCancel(task.ErrClaimLost) })
+						// The canceled context fails the next chunk, ending the run
+						dexcomClient.EXPECT().GetAlerts(matchContext(), firstChunkEndTime, secondChunkEndTime, matchNotNil()).DoAndReturn(mockDexcomClientGetData[dexcom.AlertsResponse](nil, nil, context.Canceled))
+						taskRunner.Run(claimContext)
+						assertTaskState(task.TaskStateRunning)
+						Expect(dataSrc.State).To(Equal(dataSource.StateConnected))
+						Expect(dataSrc.HasError()).To(BeFalse())
+						logger.AssertWarn("Skipped updating data source and task because the task claim was lost")
+					})
 				})
 			})
 		})

task/queue/queue.go

@@ -34,7 +34,7 @@ const (
 	DispatchTasksDelayDefault = 1 * time.Minute
 
 	// MonitorTaskDelayDefault is the default interval between checks that each in-flight task's claim is still held
-	// (the task exists and its claim token is unchanged); a run whose claim is lost is canceled with ErrClaimLost.
+	// (the task exists and its claim token is unchanged); a run whose claim is lost is canceled with task.ErrClaimLost.
 	MonitorTaskDelayDefault = 1 * time.Minute
 
 	// RunnerWatchdogGracePeriodDefault is the extra time beyond the runner timeout that the watchdog waits before
@@ -363,7 +363,7 @@ func (q *Queue) runTask(ctx context.Context, tsk *task.Task) {
 		return
 	}
 
-	// The claim context is canceled with ErrClaimLost by the task claim monitor when the task is deleted or re-claimed
+	// The claim context is canceled with task.ErrClaimLost by the task claim monitor when the task is deleted or re-claimed
 	// mid-run. Claim loss is irreversible for this run (every claim gets a fresh token), so once canceled the outcome
 	// can never be persisted.
 	claimContext, claimCancel := context.WithCancelCause(ctx)
@@ -372,7 +372,7 @@ func (q *Queue) runTask(ctx context.Context, tsk *task.Task) {
 	// Clearing the claim token marks the outcome as unpersistable, which completeTask discards. Done in a defer, after
 	// the recover below, so a panicking runner is also reconciled correctly.
 	defer func() {
-		if errors.Is(context.Cause(claimContext), ErrClaimLost) {
+		if errors.Is(context.Cause(claimContext), task.ErrClaimLost) {
 			tsk.ClaimToken = nil
 		}
 	}()
@@ -408,7 +408,7 @@ func (q *Queue) runTask(ctx context.Context, tsk *task.Task) {
 		if reason := q.monitorTaskForLostClaim(claimContext, id, claimToken); reason != nil {
 			log.LoggerFromContext(claimContext).Warnf("Task %s; canceling task run", *reason)
 			RunClaimLostTotal.WithLabelValues(typ, *reason).Inc()
-			claimCancel(ErrClaimLost)
+			claimCancel(task.ErrClaimLost)
 		}
 	}(tsk.ID, tsk.Type, *tsk.ClaimToken)
 
@@ -427,7 +427,7 @@ func (q *Queue) runTask(ctx context.Context, tsk *task.Task) {
 
 	// The claim was lost mid-run; any write-back would miss the claim token, so skip state reconciliation. The outcome
 	// is discarded during completion.
-	if errors.Is(context.Cause(claimContext), ErrClaimLost) {
+	if errors.Is(context.Cause(claimContext), task.ErrClaimLost) {
 		return
 	}
 

task/queue/runner.go

@@ -43,8 +43,8 @@ import (
 // Context: canceled (with cause) after GetRunnerTimeout, on shutdown, and on claim loss - cooperative runners select on
 // ctx.Done() and return promptly. To tell them apart, inspect context.Cause(ctx): a normal shutdown with
 // context.Canceled; a runner timeout with ErrRunnerTimeoutExceeded, and a lost claim (the task was deleted mid-run, or
-// unstuck and its claim token replaced) with ErrClaimLost (use errors.Is in all cases). After a claim loss the run's
-// outcome is discarded - no write-back can land - so return promptly and skip any remaining work. Claim loss is
+// unstuck and its claim token replaced) with task.ErrClaimLost (use errors.Is in all cases). After a claim loss the
+// run's outcome is discarded - no write-back can land - so return promptly and skip any remaining work. Claim loss is
 // detected by a periodic check (MonitorTaskDelay, 1 minute by default), not instantly. The queue cannot preempt a
 // runner that ignores cancellation; it stays blocked until it returns or the process restarts, recovered by the
 // deadline/unstick mechanism (a periodic sweep that resets tasks still running past their deadline - see
@@ -109,8 +109,3 @@ type Runner interface {
 // runner distinguishes a timeout from a shutdown with errors.Is(context.Cause(ctx), ErrRunnerTimeoutExceeded); a
 // shutdown instead cancels with context.Canceled.
 var ErrRunnerTimeoutExceeded = errors.New("task runner timeout exceeded")
-
-// ErrClaimLost is the cancellation cause set on the Run context when the run's claim on the task was lost mid-run: the
-// task was deleted, or it was unstuck and possibly re-claimed (its claim token no longer matches). A runner
-// distinguishes it with errors.Is(context.Cause(ctx), ErrClaimLost).
-var ErrClaimLost = errors.New("task claim lost")

task/task.go

@@ -348,3 +348,8 @@ func (t Tasks) Sanitize(details request.AuthDetails) error {
 	}
 	return nil
 }
+
+// ErrClaimLost is the cancellation cause set on the Run context when the run's claim on the task was lost mid-run: the
+// task was deleted, or it was unstuck and possibly re-claimed (its claim token no longer matches). A runner
+// distinguishes it with errors.Is(context.Cause(ctx), ErrClaimLost).
+var ErrClaimLost = errors.New("task claim lost")