Merge pull request #306 from kerberos-io/fix/heartbeat-offline-tus-retry

fix(cloud): prevent cameras going offline from unbounded TUS retries

machinery/src/cloud/cloud.go

@@ -231,14 +231,20 @@ func rawJSONOrEmptyArray(b []byte) json.RawMessage {
 func HandleHeartBeat(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time) {
 	log.Log.Debug("cloud.HandleHeartBeat(): started")
 
+	// Bound every heartbeat POST so a stalled connection (e.g. a saturated uplink
+	// or an unresponsive Hub/Vault) fails fast on this cycle instead of blocking
+	// the whole heartbeat loop indefinitely. A hung POST would otherwise stop all
+	// further heartbeats, and Hub marks a camera offline once its last heartbeat is
+	// older than 180s even while capture is healthy.
+	const heartbeatHTTPTimeout = 30 * time.Second
 	var client *http.Client
 	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
 		tr := &http.Transport{
 			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 		}
-		client = &http.Client{Transport: tr}
+		client = &http.Client{Transport: tr, Timeout: heartbeatHTTPTimeout}
 	} else {
-		client = &http.Client{}
+		client = &http.Client{Timeout: heartbeatHTTPTimeout}
 	}
 
 	kerberosAgentVersion := utils.VERSION

machinery/src/cloud/tus_client.go

@@ -163,6 +163,15 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	// never consume the retry budget (matching the legacy single-POST behaviour).
 	lastStatus := 0
 
+	// highWaterOffset is the furthest server-acknowledged offset observed across
+	// all attempts (via HEAD or PATCH). It lets the retry budget be refreshed only
+	// on GENUINE net forward progress. Without it, a server that keeps resetting the
+	// offset — e.g. a persistent 409 ERR_MISMATCHED_OFFSET where HEAD reports 0 again
+	// while the first chunk still "succeeds" — would refresh the budget every attempt
+	// and loop forever, wedging the upload worker on one recording and saturating the
+	// uplink.
+	highWaterOffset := int64(0)
+
 	for attempt := 0; attempt < maxAttempts; attempt++ {
 		// (1) Ensure we have an active upload URL, creating one if needed.
 		if uploadURL == "" {
@@ -196,6 +205,15 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 			continue
 		}
 
+		// The furthest offset any previous attempt reached. If this attempt pushes
+		// past it (via HEAD showing server-side progress or a successful PATCH) we made
+		// genuine net progress and may refresh the retry budget; if not, a repeated
+		// failure at the same spot must count against maxAttempts.
+		startHighWater := highWaterOffset
+		if offset > highWaterOffset {
+			highWaterOffset = offset
+		}
+
 		// (3) All bytes are present but the upload was not finalized (e.g. the
 		// completion hook failed). A completed tus upload cannot be re-finalized
 		// with another PATCH, so delete it and re-upload to force a clean finalize.
@@ -216,7 +234,6 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 		// checkpointing the offset after each one so an interruption resumes from the
 		// last completed chunk instead of re-uploading everything.
 		chunkSize := tusChunkSize()
-		progressed := false
 		patchFailed := false
 		var lastBody string
 		loggedProgressBucket := tusProgressBucket(offset, size)
@@ -244,10 +261,10 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 				patchFailed = true
 				break
 			}
-			if newOffset > offset {
-				progressed = true
-			}
 			offset = newOffset
+			if offset > highWaterOffset {
+				highWaterOffset = offset
+			}
 			lastBody = respBody
 			logTusUploadProgress(label, offset, size, &loggedProgressBucket)
 			if offset < size {
@@ -256,10 +273,13 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 			}
 		}
 		if patchFailed {
-			if progressed {
+			if highWaterOffset > startHighWater {
-				// Forward progress refreshes the retry budget: maxAttempts bounds the
+				// Genuine net progress (we advanced past the furthest point any previous
-				// number of consecutive failures, not the number of chunks needed for
+				// attempt reached) refreshes the retry budget: maxAttempts bounds the
-				// a large recording.
+				// number of consecutive *non-progressing* failures, not the number of
+				// chunks needed for a large recording. A server that keeps rejecting the
+				// same offset (no net progress, e.g. a persistent ERR_MISMATCHED_OFFSET)
+				// therefore gives up after maxAttempts instead of retrying forever.
 				attempt = -1
 			}
 			continue

machinery/src/cloud/tus_client_test.go

@@ -49,6 +49,14 @@ type fakeTus struct {
 	// after storing the bytes, simulating a failed completion hook.
 	failFinalize int
 
+	// loseProgress simulates a vault that never durably retains the in-progress
+	// upload: every PATCH is acknowledged (the response advertises the advanced
+	// offset) but the stored offset is immediately reset to 0. HEAD therefore
+	// keeps reporting 0 and the next chunk — sent at the advanced offset — is
+	// rejected with 409, reproducing the cross-replica ERR_MISMATCHED_OFFSET
+	// loop that previously wedged the agent's upload worker forever.
+	loseProgress bool
+
 	// requests records the headers of every received request (in order) so
 	// tests can assert which auth/routing headers the client sent per method.
 	requests []recordedRequest
@@ -155,6 +163,31 @@ func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusNotFound)
 			return
 		}
+		if s.loseProgress {
+			reqOffset, _ := strconv.ParseInt(r.Header.Get("Upload-Offset"), 10, 64)
+			s.mu.Lock()
+			cur := u.offset
+			if reqOffset != cur {
+				// The offset the client resumes from no longer matches what this
+				// "replica" retained, so reject like a vault returning
+				// ERR_MISMATCHED_OFFSET.
+				s.mu.Unlock()
+				w.Header().Set("Upload-Offset", strconv.FormatInt(cur, 10))
+				w.WriteHeader(http.StatusConflict)
+				return
+			}
+			n, _ := io.Copy(io.Discard, r.Body)
+			s.lastPatchBytes = n
+			s.patchSizes = append(s.patchSizes, n)
+			// Advertise progress to the client, then immediately forget it so the
+			// next chunk (sent at the advanced offset) mismatches again.
+			reported := cur + n
+			u.offset = 0
+			s.mu.Unlock()
+			w.Header().Set("Upload-Offset", strconv.FormatInt(reported, 10))
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
 		n, _ := io.Copy(io.Discard, r.Body)
 		s.mu.Lock()
 		u.offset += n
@@ -393,6 +426,61 @@ func TestUploadVaultResumable_NetworkErrorKeepsRetryBudget(t *testing.T) {
 	}
 }
 
+func TestUploadVaultResumable_MismatchedOffsetGivesUp(t *testing.T) {
+	// A vault that never durably retains the in-progress upload (offset resets to
+	// 0 between chunks) makes every resume "progress" by one chunk and then fail
+	// the next chunk with 409. Before the high-water gating fix this refreshed the
+	// retry budget every attempt and looped forever, wedging the upload worker and
+	// saturating the uplink (which starved heartbeats and reported the camera
+	// offline). The loop must now be bounded: give up after a fixed number of
+	// non-progressing attempts and report responded=true so the caller re-queues.
+	srv := newFakeTus()
+	srv.loseProgress = true
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	// Keep the between-attempt back-off tiny so the test stays fast.
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	defer func() { tusBackoffBaseDelay = oldDelay }()
+
+	// Force multiple chunks so there is always a second chunk to be rejected.
+	t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "4096")
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	withRecording(t, fileName, bytes.Repeat([]byte("m"), 12288))
+
+	done := make(chan struct{})
+	var uploaded, responded bool
+	var upErr error
+	go func() {
+		uploaded, responded, _, _, upErr = uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("resumable upload did not terminate: the retry loop is unbounded on a persistent mismatched offset")
+	}
+
+	if uploaded {
+		t.Fatal("expected uploaded=false when the vault never retains the offset")
+	}
+	if !responded {
+		t.Fatal("expected responded=true (the vault answered) so the caller re-queues the recording")
+	}
+	if upErr == nil {
+		t.Fatal("expected an error when the upload cannot complete")
+	}
+
+	// The bounded retry budget must cap the number of PATCH requests. Two PATCHes
+	// per attempt across a handful of attempts stays comfortably below this.
+	if count, _ := srv.patchCounts(); count > 50 {
+		t.Fatalf("expected a bounded number of PATCH requests, got %d (retry loop not bounded)", count)
+	}
+}
+
 func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
 	srv := newFakeTus()
 	srv.failFinalize = 1