diff --git a/machinery/src/cloud/cloud.go b/machinery/src/cloud/cloud.go index c460507..1e7ff46 100644 --- a/machinery/src/cloud/cloud.go +++ b/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 diff --git a/machinery/src/cloud/tus_client.go b/machinery/src/cloud/tus_client.go index be6f8df..53a157b 100644 --- a/machinery/src/cloud/tus_client.go +++ b/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 { - // Forward progress refreshes the retry budget: maxAttempts bounds the - // number of consecutive failures, not the number of chunks needed for - // a large recording. + if highWaterOffset > startHighWater { + // Genuine net progress (we advanced past the furthest point any previous + // attempt reached) refreshes the retry budget: maxAttempts bounds the + // 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 diff --git a/machinery/src/cloud/tus_client_test.go b/machinery/src/cloud/tus_client_test.go index d09f9f0..1a605ea 100644 --- a/machinery/src/cloud/tus_client_test.go +++ b/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