Merge pull request #300 from kerberos-io/feature/improved-cleanup-and-tus-upload-on-network-error

feature/improved-cleanup-and-tus-upload-on-network-error

README.md

@@ -203,7 +203,8 @@ Next to attaching the configuration file, it is also possible to override the co
 | `AGENT_REMOVE_AFTER_UPLOAD`                 | When enabled, recordings uploaded successfully to a storage will be removed from disk.          | "true"                         |
 | `AGENT_OFFLINE`                             | Makes sure no external connection is made.                                                      | "false"                        |
 | `AGENT_AUTO_CLEAN`                          | Cleans up the recordings directory.                                                             | "true"                         |
-| `AGENT_AUTO_CLEAN_MAX_SIZE`                 | If `AUTO_CLEAN` enabled, set the max size of the recordings directory (in MB).                  | "100"                          |
+| `AGENT_AUTO_CLEAN_MAX_SIZE`                 | If `AUTO_CLEAN` enabled, cap the recordings directory at this size (in MB). When unset/0, recordings may use the whole disk instead (see `AGENT_AUTO_CLEAN_MIN_FREE_SPACE`). | "100"                          |
+| `AGENT_AUTO_CLEAN_MIN_FREE_SPACE`           | When `AUTO_CLEAN` is enabled and no `MAX_SIZE` is set, keep at least this much free space (in MB) on the recordings disk before deleting the oldest (already-uploaded first) recordings. Defaults to 5% of the disk. | ""                             |
 | `AGENT_TIME`                                | Enable the timetable for Kerberos Agent                                                         | "false"                        |
 | `AGENT_TIMETABLE`                           | A (weekly) time table to specify when to make recordings "start1,end1,start2,end2;start1..      | ""                             |
 | `AGENT_REGION_POLYGON`                      | A single polygon set for motion detection: "x1,y1;x2,y2;x3,y3;...                               | ""                             |

machinery/src/capture/cleanup_test.go

@@ -0,0 +1,219 @@
+package capture
+
+import (
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// writeRecording creates a file under recordingsDir and sets its modtime so the
+// tests can control the "oldest" ordering deterministically.
+func writeRecording(t *testing.T, recordingsDir, name string, ageMinutes int) {
+	t.Helper()
+	full := filepath.Join(recordingsDir, name)
+	if err := os.WriteFile(full, []byte("data"), 0o644); err != nil {
+		t.Fatalf("write recording %s: %v", name, err)
+	}
+	mod := time.Now().Add(-time.Duration(ageMinutes) * time.Minute)
+	if err := os.Chtimes(full, mod, mod); err != nil {
+		t.Fatalf("chtimes %s: %v", name, err)
+	}
+}
+
+// markPending creates the upload marker in cloudDir for the given recording,
+// marking it as still queued for upload.
+func markPending(t *testing.T, cloudDir, name string) {
+	t.Helper()
+	if err := os.WriteFile(filepath.Join(cloudDir, name), nil, 0o644); err != nil {
+		t.Fatalf("write marker %s: %v", name, err)
+	}
+}
+
+func newCleanupDirs(t *testing.T) (string, string) {
+	t.Helper()
+	base := t.TempDir()
+	recordingsDir := filepath.Join(base, "data", "recordings")
+	cloudDir := filepath.Join(base, "data", "cloud")
+	if err := os.MkdirAll(recordingsDir, 0o755); err != nil {
+		t.Fatalf("mkdir recordings: %v", err)
+	}
+	if err := os.MkdirAll(cloudDir, 0o755); err != nil {
+		t.Fatalf("mkdir cloud: %v", err)
+	}
+	return recordingsDir, cloudDir
+}
+
+// The core regression: when the oldest recording is still pending upload but a
+// newer one has already been uploaded, cleanup must delete the uploaded (safe)
+// one and leave the pending recording on disk so it can still be uploaded.
+func TestPickRecordingToCleanup_PrefersUploaded(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	// oldest is still pending upload (marker present).
+	writeRecording(t, recordingsDir, "oldest_pending.mp4", 30)
+	markPending(t, cloudDir, "oldest_pending.mp4")
+	// newer one has already been uploaded (no marker).
+	writeRecording(t, recordingsDir, "newer_uploaded.mp4", 10)
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if pending {
+		t.Fatalf("expected a safe (already-uploaded) deletion, got pending=true")
+	}
+	if name != "newer_uploaded.mp4" {
+		t.Fatalf("cleanup picked %q, want the uploaded recording newer_uploaded.mp4", name)
+	}
+}
+
+// Among several already-uploaded recordings, the oldest uploaded one is chosen.
+func TestPickRecordingToCleanup_OldestUploadedFirst(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	writeRecording(t, recordingsDir, "old_uploaded.mp4", 40)
+	writeRecording(t, recordingsDir, "mid_uploaded.mp4", 20)
+	// pending one must be ignored even though it is not the oldest.
+	writeRecording(t, recordingsDir, "pending.mp4", 30)
+	markPending(t, cloudDir, "pending.mp4")
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if pending {
+		t.Fatalf("expected pending=false, got true")
+	}
+	if name != "old_uploaded.mp4" {
+		t.Fatalf("cleanup picked %q, want old_uploaded.mp4", name)
+	}
+}
+
+// Last resort: when every recording is still pending upload, cleanup returns the
+// oldest one with pending=true so the caller can drop it (and its marker) to keep
+// the disk bounded.
+func TestPickRecordingToCleanup_AllPendingFallsBackToOldest(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	writeRecording(t, recordingsDir, "a_old.mp4", 50)
+	markPending(t, cloudDir, "a_old.mp4")
+	writeRecording(t, recordingsDir, "b_new.mp4", 5)
+	markPending(t, cloudDir, "b_new.mp4")
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !pending {
+		t.Fatalf("expected pending=true when every recording is queued for upload")
+	}
+	if name != "a_old.mp4" {
+		t.Fatalf("cleanup picked %q, want the oldest pending a_old.mp4", name)
+	}
+}
+
+// An empty recordings directory yields os.ErrNotExist so the caller does nothing.
+func TestPickRecordingToCleanup_Empty(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	if _, _, err := pickRecordingToCleanup(recordingsDir, cloudDir); err != os.ErrNotExist {
+		t.Fatalf("expected os.ErrNotExist for an empty directory, got %v", err)
+	}
+}
+
+// writeSizedRecording writes a recording of an exact byte size so tests can
+// exercise the megabyte-based directory-cap threshold.
+func writeSizedRecording(t *testing.T, dir, name string, size int) {
+	t.Helper()
+	if err := os.WriteFile(filepath.Join(dir, name), make([]byte, size), 0o644); err != nil {
+		t.Fatalf("write sized recording %s: %v", name, err)
+	}
+}
+
+// When AGENT_AUTO_CLEAN_MAX_SIZE (MaxDirectorySize) is set, cleanup triggers once
+// the recordings directory grows past that many megabytes.
+func TestRecordingsNeedCleanup_FixedCap(t *testing.T) {
+	recordingsDir, _ := newCleanupDirs(t)
+	// ~2 MB of recordings on disk.
+	writeSizedRecording(t, recordingsDir, "big.mp4", 2*1000*1000)
+
+	over := &models.Configuration{Config: models.Config{MaxDirectorySize: 1}}
+	need, err := recordingsNeedCleanup(recordingsDir, over)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !need {
+		t.Fatalf("expected cleanup when 2MB of recordings exceed the 1MB cap")
+	}
+
+	under := &models.Configuration{Config: models.Config{MaxDirectorySize: 100}}
+	need, err = recordingsNeedCleanup(recordingsDir, under)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if need {
+		t.Fatalf("expected no cleanup when 2MB of recordings stay under the 100MB cap")
+	}
+}
+
+// With no fixed cap (the default), cleanup is driven by the free space left on
+// the recordings filesystem versus the reserve.
+func TestRecordingsNeedCleanup_DefaultDiskReserve(t *testing.T) {
+	if runtime.GOOS != "linux" {
+		t.Skip("disk usage stats are only implemented on linux")
+	}
+	recordingsDir, _ := newCleanupDirs(t)
+
+	totalMB, availableMB, err := diskUsageMB(recordingsDir)
+	if err != nil {
+		t.Fatalf("diskUsageMB: %v", err)
+	}
+	if totalMB <= 0 || availableMB <= 0 {
+		t.Skipf("unexpected disk stats total=%dMB available=%dMB", totalMB, availableMB)
+	}
+
+	// A reserve larger than the whole disk means free space is always below it.
+	over := &models.Configuration{Config: models.Config{MinFreeSpace: totalMB + availableMB}}
+	need, err := recordingsNeedCleanup(recordingsDir, over)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !need {
+		t.Fatalf("expected cleanup when free space (%dMB) is below the reserve", availableMB)
+	}
+
+	// A 1 MB reserve leaves plenty of free space, so nothing should be cleaned.
+	under := &models.Configuration{Config: models.Config{MinFreeSpace: 1}}
+	need, err = recordingsNeedCleanup(recordingsDir, under)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if need {
+		t.Fatalf("expected no cleanup when free space (%dMB) exceeds the 1MB reserve", availableMB)
+	}
+}
+
+// The default 5% reserve must never truncate to 0MB on small disks, otherwise
+// cleanup would only trigger once the disk is completely full.
+func TestDefaultReserveMB(t *testing.T) {
+	cases := []struct {
+		totalMB int64
+		want    int64
+	}{
+		{totalMB: 0, want: 1},   // no/unknown disk size still reserves 1MB
+		{totalMB: 10, want: 1},  // 5% = 0MB -> floored to 1MB
+		{totalMB: 19, want: 1},  // 5% = 0MB -> floored to 1MB
+		{totalMB: 20, want: 1},  // 5% = exactly 1MB
+		{totalMB: 100, want: 5}, // 5% = 5MB
+		{totalMB: 1000, want: 50},
+	}
+	for _, c := range cases {
+		if got := defaultReserveMB(c.totalMB); got != c.want {
+			t.Errorf("defaultReserveMB(%d) = %d, want %d", c.totalMB, got, c.want)
+		}
+	}
+}

machinery/src/capture/disk_linux.go

@@ -0,0 +1,23 @@
+//go:build linux
+
+package capture
+
+import "syscall"
+
+// diskUsageMB returns the total capacity and the currently available space (both
+// in megabytes, decimal) of the filesystem that contains path. Auto-clean uses
+// it to default its cleanup threshold to the real disk capacity instead of a
+// fixed size, so recordings can grow to fill the disk while keeping a reserve
+// free. Linux is the agent's deployment target (amd64/arm64 containers).
+func diskUsageMB(path string) (totalMB int64, availableMB int64, err error) {
+	var stat syscall.Statfs_t
+	if err = syscall.Statfs(path, &stat); err != nil {
+		return 0, 0, err
+	}
+	blockSize := int64(stat.Bsize)
+	totalMB = int64(stat.Blocks) * blockSize / 1000 / 1000
+	// Bavail is the free space available to unprivileged users, which is the
+	// space we can actually keep writing recordings into.
+	availableMB = int64(stat.Bavail) * blockSize / 1000 / 1000
+	return totalMB, availableMB, nil
+}

machinery/src/capture/disk_other.go

@@ -0,0 +1,13 @@
+//go:build !linux
+
+package capture
+
+import "errors"
+
+// diskUsageMB is only implemented on Linux (the agent's deployment target). On
+// other platforms (e.g. local macOS/Windows dev builds) auto-clean falls back to
+// its historical fixed-size directory cap, so this reports the capability as
+// unavailable.
+func diskUsageMB(path string) (totalMB int64, availableMB int64, err error) {
+	return 0, 0, errors.New("disk usage stats are not supported on this platform")
+}

machinery/src/capture/main.go

@@ -22,36 +22,170 @@ import (
 
 func CleanupRecordingDirectory(configDirectory string, configuration *models.Configuration) {
 	autoClean := configuration.Config.AutoClean
-	if autoClean == "true" {
-		maxSize := configuration.Config.MaxDirectorySize
-		if maxSize == 0 {
-			maxSize = 300
+	if autoClean != "true" {
+		log.Log.Info("HandleRecordStream: Autoclean disabled, nothing to do here.")
+		return
+	}
+
+	recordingsDirectory := configDirectory + "/data/recordings"
+	cloudDirectory := configDirectory + "/data/cloud"
+
+	// Decide whether we still need to free up space. See recordingsNeedCleanup
+	// for the two modes: an explicit fixed directory cap
+	// (AGENT_AUTO_CLEAN_MAX_SIZE) or, by default, letting recordings use the whole
+	// disk while keeping a free-space reserve.
+	needsCleanup, err := recordingsNeedCleanup(recordingsDirectory, configuration)
+	if err != nil {
+		log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
+		return
+	}
+	if !needsCleanup {
+		return
+	}
+
+	// Remove the oldest recording, but PREFER recordings that have already been
+	// uploaded (i.e. no longer have a pending marker in data/cloud). This stops
+	// auto-clean from deleting recordings that are still queued for upload. That
+	// previously caused silent data loss: during a network outage the upload
+	// backlog grows, cleanup deletes the oldest (still un-uploaded) recording to
+	// stay under MaxDirectorySize, and when connectivity returns the upload loop
+	// finds the marker but the file is gone -> the recording is dropped and never
+	// reaches the vault.
+	//
+	// Only when EVERY recording on disk is still pending upload do we fall back to
+	// deleting the oldest pending one, as a last resort to keep the disk bounded
+	// (otherwise a long outage would fill the disk and stop new recordings).
+	name, pending, err := pickRecordingToCleanup(recordingsDirectory, cloudDirectory)
+	if err != nil {
+		log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
+		return
+	}
+
+	if err := os.Remove(recordingsDirectory + "/" + name); err != nil {
+		log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
+		return
+	}
+
+	if pending {
+		// Data-loss event: the whole recordings directory is an un-uploaded
+		// backlog (e.g. a prolonged network outage), so we had to drop a recording
+		// that was never uploaded to keep recording new footage. Also remove the
+		// now-dangling upload marker so the upload loop doesn't keep trying to
+		// upload a file that no longer exists.
+		log.Log.Warning("HandleRecordStream: removed oldest recording as part of cleanup, but it was STILL PENDING UPLOAD (disk full of un-uploaded recordings) - " + recordingsDirectory + "/" + name)
+		if err := os.Remove(cloudDirectory + "/" + name); err != nil && !os.IsNotExist(err) {
+			log.Log.Info("HandleRecordStream: could not remove dangling upload marker " + name + ", " + err.Error())
 		}
-		// Total size of the recording directory.
-		recordingsDirectory := configDirectory + "/data/recordings"
+	} else {
+		log.Log.Info("HandleRecordStream: removed oldest file as part of cleanup - " + recordingsDirectory + "/" + name)
+	}
+}
+
+// recordingsNeedCleanup reports whether auto-clean should free up space in the
+// recordings directory. There are two modes:
+//
+//   - AGENT_AUTO_CLEAN_MAX_SIZE (MaxDirectorySize, MB) set: cap the size of the
+//     recordings directory itself (the historical behaviour).
+//   - MaxDirectorySize == 0 (the default): recordings may use the WHOLE disk.
+//     Cleanup only triggers once the free space on the recordings filesystem
+//     drops to/below a reserve. The reserve is AGENT_AUTO_CLEAN_MIN_FREE_SPACE
+//     (MinFreeSpace, MB) when set, otherwise 5% of the disk's total capacity.
+//
+// If disk stats can't be read (e.g. non-Linux dev builds) it falls back to the
+// historical fixed 300 MB directory cap so behaviour stays bounded.
+func recordingsNeedCleanup(recordingsDirectory string, configuration *models.Configuration) (bool, error) {
+	maxSize := configuration.Config.MaxDirectorySize
+
+	// Explicit fixed cap on the recordings directory size.
+	if maxSize > 0 {
 		size, err := utils.DirSize(recordingsDirectory)
-		if err == nil {
-			sizeInMB := size / 1000 / 1000
-			if sizeInMB >= maxSize {
-				// Remove the oldest recording
-				oldestFile, err := utils.FindOldestFile(recordingsDirectory)
-				if err == nil {
-					err := os.Remove(recordingsDirectory + "/" + oldestFile.Name())
-					log.Log.Info("HandleRecordStream: removed oldest file as part of cleanup - " + recordingsDirectory + "/" + oldestFile.Name())
-					if err != nil {
-						log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
-					}
-				} else {
-					log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
-				}
-			}
-		} else {
-			log.Log.Info("HandleRecordStream: something went wrong, " + err.Error())
+		if err != nil {
+			return false, err
+		}
+		return size/1000/1000 >= maxSize, nil
+	}
+
+	// Default: allow recordings to use the full disk, keeping a reserve free.
+	totalMB, availableMB, err := diskUsageMB(recordingsDirectory)
+	if err != nil {
+		// Disk stats unavailable: fall back to the historical 300 MB cap.
+		size, derr := utils.DirSize(recordingsDirectory)
+		if derr != nil {
+			return false, derr
+		}
+		return size/1000/1000 >= 300, nil
+	}
+
+	reserveMB := configuration.Config.MinFreeSpace
+	if reserveMB <= 0 {
+		reserveMB = defaultReserveMB(totalMB)
+	}
+
+	return availableMB <= reserveMB, nil
+}
+
+// defaultReserveMB returns the free-space reserve (MB) to keep on the recordings
+// disk when AGENT_AUTO_CLEAN_MIN_FREE_SPACE is not set: 5% of the disk total,
+// but never below 1MB. On very small disks 5% truncates to 0MB, which would
+// disable the reserve entirely (cleanup only once availableMB <= 0), so we floor
+// it at 1MB to preserve the intended "keep some space free" behaviour.
+func defaultReserveMB(totalMB int64) int64 {
+	reserveMB := totalMB * 5 / 100
+	if reserveMB < 1 {
+		reserveMB = 1
+	}
+	return reserveMB
+}
+
+// pickRecordingToCleanup chooses which recording to delete to free space in the
+// recordings directory. It returns the oldest recording that has already been
+// uploaded (no pending marker with the same name in cloudDirectory). Only when
+// every recording is still pending upload does it return the oldest recording
+// overall with pending=true, signalling the caller that it is about to drop an
+// un-uploaded recording as a last resort.
+func pickRecordingToCleanup(recordingsDirectory, cloudDirectory string) (string, bool, error) {
+	entries, err := os.ReadDir(recordingsDirectory)
+	if err != nil {
+		return "", false, err
+	}
+
+	var oldestSafeName, oldestAnyName string
+	var oldestSafeTime, oldestAnyTime time.Time
+
+	for _, entry := range entries {
+		if entry.IsDir() {
+			continue
+		}
+		info, err := entry.Info()
+		if err != nil || !info.Mode().IsRegular() {
+			continue
+		}
+		modTime := info.ModTime()
+
+		if oldestAnyName == "" || modTime.Before(oldestAnyTime) {
+			oldestAnyName = entry.Name()
+			oldestAnyTime = modTime
 		}
 
-	} else {
-		log.Log.Info("HandleRecordStream: Autoclean disabled, nothing to do here.")
+		// A recording is still pending upload if a marker with the same name
+		// exists in the cloud directory. Skip those when picking a safe candidate.
+		if _, statErr := os.Stat(cloudDirectory + "/" + entry.Name()); statErr == nil {
+			continue
+		}
+
+		if oldestSafeName == "" || modTime.Before(oldestSafeTime) {
+			oldestSafeName = entry.Name()
+			oldestSafeTime = modTime
+		}
 	}
+
+	if oldestSafeName != "" {
+		return oldestSafeName, false, nil
+	}
+	if oldestAnyName != "" {
+		return oldestAnyName, true, nil
+	}
+	return "", false, os.ErrNotExist
 }
 
 func HandleRecordStream(queue *packets.Queue, configDirectory string, configuration *models.Configuration, communication *models.Communication, rtspClient RTSPClient) {

machinery/src/cloud/kerberos_vault.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"strconv"
 	"time"
 
 	"github.com/kerberos-io/agent/machinery/src/log"
@@ -199,17 +200,59 @@ func setVaultHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey,
 }
 
 // newVaultHTTPClient builds an HTTP client honouring the AGENT_TLS_INSECURE
-// escape hatch. A timeout of 0 disables the client-level timeout, which is
-// required for streaming large upload bodies.
+// escape hatch. A timeout of 0 disables the *overall* client timeout, which is
+// required for streaming large upload bodies without capping the total transfer
+// time. Transport-level timeouts are still applied so that a lost network
+// connection (for example the internet being disconnected) fails reasonably
+// fast and the upload is retried, instead of the request hanging until the OS
+// TCP timeout (which can be many minutes) and blocking the whole upload loop.
 func newVaultHTTPClient(timeout time.Duration) *http.Client {
-	client := &http.Client{}
+	// Start from a clone of the default transport so we keep its sane dial and
+	// TLS-handshake timeouts, connection pooling and HTTP/2 support even when the
+	// AGENT_TLS_INSECURE escape hatch is enabled (a bare http.Transport would have
+	// no dial/handshake timeouts at all).
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+
+	// ResponseHeaderTimeout bounds how long we wait for the vault's response
+	// headers *after* the request body has been fully written. It does not limit
+	// the time spent streaming the (potentially large) upload body, so big
+	// recordings still upload fine, but a vault/network that disappears while we
+	// wait for the acknowledgement is detected and the upload is retried instead
+	// of hanging indefinitely.
+	transport.ResponseHeaderTimeout = vaultResponseHeaderTimeout()
+
 	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-		client.Transport = &http.Transport{
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+		if transport.TLSClientConfig == nil {
+			transport.TLSClientConfig = &tls.Config{}
 		}
+		transport.TLSClientConfig.InsecureSkipVerify = true
 	}
+
+	client := &http.Client{Transport: transport}
 	if timeout > 0 {
 		client.Timeout = timeout
 	}
 	return client
 }
+
+// vaultResponseHeaderTimeout returns the maximum time to wait for a vault's
+// response headers after the request body has been written. It defaults to 5
+// minutes — generous enough for the vault to persist/finalize a chunk or a full
+// recording to its storage provider — and can be tuned with the
+// AGENT_VAULT_RESPONSE_HEADER_TIMEOUT_SECONDS environment variable. A value of 0
+// (or a negative/invalid value) disables the timeout.
+func vaultResponseHeaderTimeout() time.Duration {
+	const def = 5 * time.Minute
+	v := os.Getenv("AGENT_VAULT_RESPONSE_HEADER_TIMEOUT_SECONDS")
+	if v == "" {
+		return def
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return def
+	}
+	if n <= 0 {
+		return 0
+	}
+	return time.Duration(n) * time.Second
+}

machinery/src/cloud/tus_client.go

@@ -154,10 +154,20 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	const maxAttempts = 4
 	restartedAfterComplete := false
 
+	// lastStatus holds the HTTP status code of the most recent tus request. A
+	// value of 0 means the request failed at the transport level (no HTTP
+	// response at all, e.g. the vault was unreachable or the connection dropped
+	// because the internet went down). It lets the final "gave up" return report
+	// whether the vault actually answered, so the caller only advances its
+	// retry/back-off policy on a definitive response and transient network errors
+	// never consume the retry budget (matching the legacy single-POST behaviour).
+	lastStatus := 0
+
 	for attempt := 0; attempt < maxAttempts; attempt++ {
 		// (1) Ensure we have an active upload URL, creating one if needed.
 		if uploadURL == "" {
 			created, status, cerr := tusCreate(client, baseURL, size, metadata, setHeaders, fileName)
+			lastStatus = status
 			if cerr != nil {
 				if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented {
 					// The vault does not implement tus; let the caller fall back.
@@ -173,6 +183,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 
 		// (2) Query the current server-side offset.
 		offset, status, herr := tusHead(client, uploadURL, setHeaders)
+		lastStatus = status
 		if herr != nil {
 			if status == http.StatusNotFound || status == http.StatusGone {
 				// The upload expired/was removed server-side; start over.
@@ -220,6 +231,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 				patchLen = chunkSize
 			}
 			newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, patchLen, file, setHeaders)
+			lastStatus = status
 			if perr != nil {
 				if status >= 400 {
 					// Definitive rejection (e.g. provider push failed during finalize).
@@ -258,7 +270,13 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 		return true, true, true, lastBody, nil
 	}
 
-	return false, true, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
+	// Every attempt failed. Only report responded=true when the vault actually
+	// answered on the last attempt (lastStatus > 0). If every attempt failed at
+	// the transport level (lastStatus == 0, e.g. the internet was disconnected),
+	// report responded=false so the caller keeps the recording queued and retries
+	// later instead of consuming its retry budget and entering the long back-off
+	// timeout.
+	return false, lastStatus > 0, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
 }
 
 // uploadVaultResumable uploads a recording directly to a Kerberos Vault using
@@ -535,10 +553,15 @@ func removeTusResumeState(path string) {
 	_ = os.Remove(path)
 }
 
+// tusBackoffBaseDelay is the base delay used by tusBackoff for the exponential
+// back-off between resume attempts. It is a package variable (rather than a
+// constant) so tests can shrink it to keep them fast.
+var tusBackoffBaseDelay = 500 * time.Millisecond
+
 // tusBackoff sleeps for an exponentially increasing duration (capped) between
 // resume attempts to avoid hammering a temporarily unavailable vault.
 func tusBackoff(attempt int) {
-	delay := time.Duration(500*(1<<uint(attempt))) * time.Millisecond
+	delay := tusBackoffBaseDelay * time.Duration(1<<uint(attempt))
 	if delay > 3*time.Second {
 		delay = 3 * time.Second
 	}

machinery/src/cloud/tus_client_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -13,6 +14,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
@@ -350,6 +352,47 @@ func TestUploadVaultResumable_Unsupported(t *testing.T) {
 	}
 }
 
+// TestUploadVaultResumable_NetworkErrorKeepsRetryBudget verifies that when the
+// vault is unreachable (mimicking the internet being disconnected) the resumable
+// upload reports responded=false. That is what stops the caller
+// (UploadKerberosVault) from consuming its retry budget and entering the long
+// back-off timeout on a transient network outage, so the recording keeps being
+// retried until connectivity returns.
+func TestUploadVaultResumable_NetworkErrorKeepsRetryBudget(t *testing.T) {
+	// Bind then immediately release a loopback port so every connection to it is
+	// refused, producing a transport-level error (no HTTP response).
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	addr := ln.Addr().String()
+	if cerr := ln.Close(); cerr != nil {
+		t.Fatalf("close listener: %v", cerr)
+	}
+
+	// Keep the between-attempt back-off tiny so the test stays fast.
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	defer func() { tusBackoffBaseDelay = oldDelay }()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	withRecording(t, fileName, bytes.Repeat([]byte("n"), 2048))
+
+	uploaded, responded, supported, _, err := uploadVaultResumable(testVault("http://"+addr), "pk", "dev", fileName, "test", "primary")
+	if uploaded {
+		t.Fatal("expected uploaded=false when the vault is unreachable")
+	}
+	if !supported {
+		t.Fatal("a transport error is not a missing tus endpoint; expected supported=true")
+	}
+	if responded {
+		t.Fatal("expected responded=false for a pure network error so the retry budget is preserved")
+	}
+	if err == nil {
+		t.Fatal("expected an error when the vault is unreachable")
+	}
+}
+
 func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
 	srv := newFakeTus()
 	srv.failFinalize = 1

machinery/src/config/main.go

@@ -327,6 +327,12 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 					configuration.Config.MaxDirectorySize = size
 				}
 				break
+			case "AGENT_AUTO_CLEAN_MIN_FREE_SPACE":
+				size, err := strconv.ParseInt(value, 10, 64)
+				if err == nil {
+					configuration.Config.MinFreeSpace = size
+				}
+				break
 
 			/* Camera configuration */
 			case "AGENT_CAPTURE_IPCAMERA_RTSP":

machinery/src/models/config.go

@@ -21,6 +21,7 @@ type Config struct {
 	AutoClean               string       `json:"auto_clean"`
 	RemoveAfterUpload       string       `json:"remove_after_upload"`
 	MaxDirectorySize        int64        `json:"max_directory_size"`
+	MinFreeSpace            int64        `json:"min_free_space,omitempty"`
 	Timezone                string       `json:"timezone"`
 	Capture                 Capture      `json:"capture"`
 	Timetable               []*Timetable `json:"timetable"`