Merge pull request #344 from kerberos-io/fix/tus-minio-upload-empty-region

fix/tus-minio-upload-empty-region?expand=1

README.md

@@ -99,6 +99,9 @@ back to the legacy filename format and configured Agent key. A structured
 checkpoint takes precedence over both the filename and current configured key;
 an invalid or unreadable checkpoint is reported as an upload error, not silently
 replaced with filename-derived values.
+For older structured checkpoints, missing camera name, region, and change count
+are recovered from a validated legacy filename. Populated checkpoint fields
+take precedence, and the checkpoint timestamp and device key remain authoritative.
 
 The shorter filename cannot be decoded by legacy filename-only consumers.
 Integrations must preserve structured recording metadata, including the device
@@ -461,6 +464,15 @@ Smaller chunks provide more frequent resumable checkpoints but create more HTTP
 requests. Larger chunks reduce request overhead but require more data to be
 retransmitted when a request fails.
 
+An HTTP 500 during resumable creation is a server/provider failure, not proof
+that the recording filename is invalid. Creation errors include up to 4 KiB of
+the server response, and the final retry error retains the latest failure cause.
+Check the corresponding Hub/Vault logs and storage-provider configuration.
+As a temporary diagnostic workaround, set `AGENT_DISABLE_RESUMABLE_UPLOAD=true`
+and restart Agent to use ordinary uploads. This does not change the recording
+filename or remove its structured metadata, but interrupted uploads restart
+from the beginning. Remove the override after fixing the provider issue.
+
 ### Custom metadata headers for Kerberos Vault
 
 Use `AGENT_KERBEROSVAULT_CUSTOM_HEADERS` to attach your own metadata (for

machinery/src/cloud/s3.go

@@ -146,6 +146,13 @@ func s3RecordingUploadDetails(config models.Config, fileName, publicKey string)
 		deviceKey = metadata.DeviceKey
 		startRecording = metadata.Timestamp
 		userMetadata["event-instancename"] = config.Name
+		if metadata.DeviceName == "" || metadata.RegionCoordinates == "" || metadata.NumberOfChanges == "" {
+			if description, ok := s3LegacyRecordingDescription(fileName); ok {
+				for key, value := range description {
+					userMetadata[key] = value
+				}
+			}
+		}
 		if metadata.DeviceName != "" {
 			userMetadata["event-instancename"] = metadata.DeviceName
 		}
@@ -203,3 +210,47 @@ func s3RecordingUploadDetails(config models.Config, fileName, publicKey string)
 	userMetadata["productid"] = deviceKey
 	return s3RecordingObjectName(config.S3.Username, deviceKey, startRecording), userMetadata, nil
 }
+
+func s3LegacyRecordingDescription(fileName string) (map[string]string, bool) {
+	base := filepath.Base(fileName)
+	if filepath.Ext(base) != ".mp4" {
+		return nil, false
+	}
+	parts := strings.Split(strings.TrimSuffix(base, ".mp4"), "_")
+	if len(parts) != 6 || parts[2] == "" {
+		return nil, false
+	}
+	seconds, err := strconv.ParseInt(parts[0], 10, 64)
+	if err != nil || seconds <= 0 || seconds > math.MaxInt64/1000 {
+		return nil, false
+	}
+	digits, fraction, found := strings.Cut(parts[1], "-")
+	precision, err := strconv.Atoi(digits)
+	if !found || err != nil || precision < 1 || precision > 6 || len(fraction) != precision {
+		return nil, false
+	}
+	if _, err := strconv.ParseUint(fraction, 10, 64); err != nil {
+		return nil, false
+	}
+	coordinates := strings.Split(parts[3], "-")
+	if len(coordinates) != 4 {
+		return nil, false
+	}
+	for _, coordinate := range coordinates {
+		if _, err := strconv.ParseUint(coordinate, 10, 64); err != nil {
+			return nil, false
+		}
+	}
+	changes, err := strconv.ParseInt(parts[4], 10, 64)
+	if err != nil || changes < -1 {
+		return nil, false
+	}
+	if _, err := strconv.ParseUint(parts[5], 10, 64); err != nil {
+		return nil, false
+	}
+	return map[string]string{
+		"event-instancename":      parts[2],
+		"event-regioncoordinates": parts[3],
+		"event-numberofchanges":   parts[4],
+	}, true
+}

machinery/src/cloud/s3_test.go

@@ -70,6 +70,109 @@ func TestS3RecordingUploadDetailsUsesCheckpoint(t *testing.T) {
 	}
 }
 
+func TestS3RecordingUploadDetailsBackfillsLegacyCheckpointDescriptions(t *testing.T) {
+	const legacyName = "1564859471_6-474162_recorded-camera_1-2-3-4_1153_27.mp4"
+	tests := []struct {
+		name        string
+		fileName    string
+		camera      string
+		region      string
+		changes     string
+		wantCamera  string
+		wantRegion  string
+		wantChanges string
+	}{
+		{"pre-upgrade checkpoint", legacyName, "", "", "", "recorded-camera", "1-2-3-4", "1153"},
+		{"checkpoint camera", legacyName, "snapshot-camera", "", "", "snapshot-camera", "1-2-3-4", "1153"},
+		{"checkpoint region", legacyName, "", "5-6-7-8", "", "recorded-camera", "5-6-7-8", "1153"},
+		{"checkpoint zero changes", legacyName, "", "", "0", "recorded-camera", "1-2-3-4", "0"},
+		{"dated reference", "2019/08/03/" + legacyName, "", "", "", "recorded-camera", "1-2-3-4", "1153"},
+		{"continuous recording", "1564859471_3-474_recorded-camera_0-0-0-0_-1_27.mp4", "", "", "", "recorded-camera", "0-0-0-0", "-1"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			withRecording(t, filepath.Base(tt.fileName), []byte("recording"))
+			metadata := models.RecordingUploadMetadata{
+				FileName:          filepath.Base(tt.fileName),
+				DeviceKey:         "checkpoint-key",
+				Timestamp:         1767355200123,
+				Duration:          20452,
+				FPS:               29.97,
+				Encrypted:         true,
+				DeviceName:        tt.camera,
+				RegionCoordinates: tt.region,
+				NumberOfChanges:   tt.changes,
+			}
+			payload, err := json.Marshal(metadata)
+			if err != nil {
+				t.Fatal(err)
+			}
+			withQueuedRecordingFPS(t, tt.fileName, string(payload))
+			config := models.Config{Key: "current-key", Name: "current-camera", S3: &models.S3{Username: "user"}}
+			objectName, got, err := s3RecordingUploadDetails(config, tt.fileName, "public-key")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if want := "user/2026/01/02/1767355200123-checkpoint-key.mp4"; objectName != want {
+				t.Fatalf("object name = %q, want %q", objectName, want)
+			}
+			want := map[string]string{
+				"publickey":               "public-key",
+				"uploadtime":              "now",
+				"event-instancename":      tt.wantCamera,
+				"event-regioncoordinates": tt.wantRegion,
+				"event-numberofchanges":   tt.wantChanges,
+				"event-microseconds":      "123000",
+				"event-timestamp":         "1767355200",
+				"productid":               "checkpoint-key",
+				"duration":                "20452",
+				"fps":                     "29.97",
+				"encrypted":               "true",
+			}
+			if !reflect.DeepEqual(got, want) {
+				t.Fatalf("metadata = %#v, want %#v", got, want)
+			}
+		})
+	}
+}
+
+func TestS3RecordingUploadDetailsDoesNotBackfillInvalidLegacyNames(t *testing.T) {
+	for _, fileName := range []string{
+		"1767355200123-checkpoint-key.mp4",
+		"invalid_6-474162_camera_0-0-0-0_1153_27.mp4",
+		"1564859471_6-invalid_camera_0-0-0-0_1153_27.mp4",
+		"1564859471_3-474162_camera_0-0-0-0_1153_27.mp4",
+		"1564859471_6-474162__0-0-0-0_1153_27.mp4",
+		"1564859471_6-474162_camera_invalid_1153_27.mp4",
+		"1564859471_6-474162_camera_0-0-0-0_invalid_27.mp4",
+		"1564859471_6-474162_camera_0-0-0-0_1153_invalid.mp4",
+		"1564859471_6-474162_camera_0-0-0-0_1153_27_extra.mp4",
+	} {
+		t.Run(fileName, func(t *testing.T) {
+			withRecording(t, fileName, []byte("recording"))
+			metadata := models.RecordingUploadMetadata{
+				FileName: fileName, DeviceKey: "checkpoint-key", Timestamp: 1767355200123, Duration: 20452,
+			}
+			payload, err := json.Marshal(metadata)
+			if err != nil {
+				t.Fatal(err)
+			}
+			withQueuedRecordingFPS(t, fileName, string(payload))
+			config := models.Config{Key: "current-key", Name: "current-camera", S3: &models.S3{Username: "user"}}
+			objectName, got, err := s3RecordingUploadDetails(config, fileName, "public-key")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if objectName != "user/2026/01/02/1767355200123-checkpoint-key.mp4" ||
+				got["event-instancename"] != "current-camera" || got["event-regioncoordinates"] != "" ||
+				got["event-numberofchanges"] != "" || got["event-timestamp"] != "1767355200" ||
+				got["productid"] != "checkpoint-key" {
+				t.Fatalf("unexpected fallback for invalid legacy name: %q, %#v", objectName, got)
+			}
+		})
+	}
+}
+
 func TestS3RecordingUploadDetailsLegacyMarkers(t *testing.T) {
 	for _, marker := range []string{"missing", "empty", "fps-only", "old-marker-name"} {
 		t.Run(marker, func(t *testing.T) {

machinery/src/cloud/tus_client.go

@@ -164,13 +164,16 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	// and loop forever, wedging the upload worker on one recording and saturating the
 	// uplink.
 	highWaterOffset := int64(0)
+	var lastErr error
 
 	for attempt := 0; attempt < maxAttempts; attempt++ {
+		lastErr = nil
 		// (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 {
+				lastErr = cerr
 				if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented {
 					// The vault does not implement tus; let the caller fall back.
 					return false, false, false, "", cerr
@@ -187,6 +190,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 		offset, status, herr := tusHead(client, uploadURL, setHeaders)
 		lastStatus = status
 		if herr != nil {
+			lastErr = herr
 			if status == http.StatusNotFound || status == http.StatusGone {
 				// The upload expired/was removed server-side; start over.
 				removeTusResumeState(sidecar)
@@ -243,6 +247,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 			newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, patchLen, file, setHeaders)
 			lastStatus = status
 			if perr != nil {
+				lastErr = perr
 				if status >= 400 {
 					// Definitive rejection (e.g. provider push failed during finalize).
 					// Re-evaluate via HEAD on the next iteration to decide retry/restart.
@@ -289,7 +294,12 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	// 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")
+	message := label + ": resumable upload did not complete after retries"
+	err = errors.New(message)
+	if lastErr != nil {
+		err = fmt.Errorf("%s: %w", message, lastErr)
+	}
+	return false, lastStatus > 0, true, "resumable upload did not complete after retries", err
 }
 
 // uploadVaultResumable uploads a recording directly to a Kerberos Vault using
@@ -395,11 +405,28 @@ func tusCreate(client *http.Client, baseURL string, size int64, metadata string,
 	if err != nil {
 		return "", 0, err
 	}
-	io.Copy(io.Discard, resp.Body)
-
 	if resp.StatusCode != http.StatusCreated {
+		const maxErrorBodyBytes = 4096
+		body, readErr := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes+1))
+		if readErr != nil {
+			return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s; read error response: %w", resp.Status, readErr)
+		}
+		truncated := len(body) > maxErrorBodyBytes
+		if truncated {
+			body = body[:maxErrorBodyBytes]
+		}
+		message := strings.TrimSpace(string(body))
+		if truncated {
+			message += " [truncated]"
+		}
+		if message != "" {
+			return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s: %q", resp.Status, message)
+		}
 		return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s", resp.Status)
 	}
+	if _, err := io.Copy(io.Discard, resp.Body); err != nil {
+		return "", resp.StatusCode, fmt.Errorf("read upload creation response: %w", err)
+	}
 	location := resp.Header.Get("Location")
 	if location == "" {
 		return "", resp.StatusCode, errors.New("missing Location header in create response")

machinery/src/cloud/tus_client_test.go

@@ -14,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -21,6 +22,72 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/utils"
 )
 
+func TestTusCreateReportsBoundedServerErrors(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		body string
+		want string
+	}{
+		{"server error", "ERR_INTERNAL_SERVER_ERROR: A region must be set when sending requests to S3.\n", "A region must be set"},
+		{"empty error", "", "500 Internal Server Error"},
+		{"large error", strings.Repeat("x", 10000), "[truncated]"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusInternalServerError)
+				_, _ = io.WriteString(w, tc.body)
+			}))
+			defer server.Close()
+			location, status, err := tusCreate(server.Client(), server.URL, 8, "",
+				func(http.Header, string) {}, "1767355200123-device-key.mp4")
+			if location != "" || status != http.StatusInternalServerError || err == nil ||
+				!strings.Contains(err.Error(), tc.want) || len(err.Error()) > 4300 {
+				t.Fatalf("creation = %q, %d, %v; want bounded server error", location, status, err)
+			}
+		})
+	}
+}
+
+func TestResumableCreateFailureRetainsServerCause(t *testing.T) {
+	previousDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = 0
+	t.Cleanup(func() { tusBackoffBaseDelay = previousDelay })
+	for _, integration := range []string{"vault", "hub"} {
+		t.Run(integration, func(t *testing.T) {
+			const fileName = "1767355200123-device-key.mp4"
+			withRecording(t, fileName, []byte("recording"))
+			var requests atomic.Int32
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				requests.Add(1)
+				if r.Method != http.MethodPost || r.URL.Path != tusUploadPath {
+					t.Errorf("unexpected request: %s %s", r.Method, r.URL.Path)
+				}
+				w.WriteHeader(http.StatusInternalServerError)
+				_, _ = io.WriteString(w, "ERR_INTERNAL_SERVER_ERROR: A region must be set when sending requests to S3.")
+			}))
+			defer server.Close()
+			var uploaded, responded, supported bool
+			var err error
+			if integration == "vault" {
+				uploaded, responded, supported, _, err = uploadVaultResumable(testVault(server.URL), "public", "device-key", fileName, "test", "primary")
+			} else {
+				uploaded, responded, supported, _, err = uploadHubResumable(testHubConfig(server.URL), fileName, "test", "hub")
+			}
+			if uploaded || !responded || !supported || err == nil ||
+				!strings.Contains(err.Error(), "resumable upload did not complete after retries") ||
+				!strings.Contains(err.Error(), "A region must be set") {
+				t.Fatalf("upload = %t/%t/%t, %v; want original server cause after retries", uploaded, responded, supported, err)
+			}
+			if requests.Load() != 4 {
+				t.Fatalf("creation attempts = %d, want 4 without legacy fallback", requests.Load())
+			}
+			if _, err := os.Stat(filepath.Join("data/recordings", fileName)); err != nil {
+				t.Fatalf("failed upload must retain the local recording: %v", err)
+			}
+		})
+	}
+}
+
 func TestUploadKerberosVaultReadsDatedLocalRecordings(t *testing.T) {
 	previousCount, previousTimeout := kstorageRetryCount, kstorageRetryTimeout
 	t.Cleanup(func() { kstorageRetryCount, kstorageRetryTimeout = previousCount, previousTimeout })

machinery/src/utils/recordings.go

@@ -107,9 +107,16 @@ func (file recordingFileInfo) Name() string {
 // ReadRecordingDirectory returns MP4s with root-relative names, preserving the
 // file information used by the existing sorting and media formatting helpers.
 func ReadRecordingDirectory(directory string) ([]os.FileInfo, error) {
+	return readRecordingDirectory(directory, filepath.WalkDir)
+}
+
+func readRecordingDirectory(directory string, walkDir func(string, fs.WalkDirFunc) error) ([]os.FileInfo, error) {
 	files := []os.FileInfo{}
-	err := filepath.WalkDir(directory, func(filePath string, entry fs.DirEntry, err error) error {
+	err := walkDir(directory, func(filePath string, entry fs.DirEntry, err error) error {
 		if err != nil {
+			if os.IsNotExist(err) && filePath != directory {
+				return nil
+			}
 			return err
 		}
 		if entry.IsDir() && entry.Name() == "decrypted" && filePath != directory {
@@ -120,6 +127,9 @@ func ReadRecordingDirectory(directory string) ([]os.FileInfo, error) {
 		}
 		info, err := entry.Info()
 		if err != nil {
+			if os.IsNotExist(err) {
+				return nil
+			}
 			return err
 		}
 		name, err := filepath.Rel(directory, filePath)

machinery/src/utils/recordings_test.go

@@ -1,7 +1,9 @@
 package utils
 
 import (
+	"errors"
 	"io"
+	"io/fs"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -202,6 +204,80 @@ func TestReadRecordingDirectoryIncludesFlatAndDatedMedia(t *testing.T) {
 	}
 }
 
+func TestReadRecordingDirectoryContinuesAfterConcurrentDeletion(t *testing.T) {
+	for _, origin := range []string{"Info", "WalkDir"} {
+		t.Run(origin, func(t *testing.T) {
+			directory := t.TempDir()
+			for _, name := range []string{"deleted.mp4", "surviving.mp4"} {
+				if err := os.WriteFile(filepath.Join(directory, name), []byte("video"), 0o644); err != nil {
+					t.Fatal(err)
+				}
+			}
+			entries, err := os.ReadDir(directory)
+			if err != nil {
+				t.Fatal(err)
+			}
+			files, err := readRecordingDirectory(directory, func(root string, visit fs.WalkDirFunc) error {
+				deleted := filepath.Join(root, entries[0].Name())
+				if err := os.Remove(deleted); err != nil {
+					t.Fatal(err)
+				}
+				var err error
+				if origin == "Info" {
+					err = visit(deleted, entries[0], nil)
+				} else {
+					err = visit(deleted, nil, &fs.PathError{Op: "lstat", Path: deleted, Err: fs.ErrNotExist})
+				}
+				if err != nil {
+					return err
+				}
+				return visit(filepath.Join(root, entries[1].Name()), entries[1], nil)
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(files) != 1 || files[0].Name() != "surviving.mp4" || files[0].Size() != 5 {
+				t.Fatalf("expected surviving recording after concurrent deletion, got %v", files)
+			}
+		})
+	}
+}
+
+type recordingDirEntryInfoError struct {
+	fs.DirEntry
+	err error
+}
+
+func (entry recordingDirEntryInfoError) Info() (fs.FileInfo, error) {
+	return nil, entry.err
+}
+
+func TestReadRecordingDirectoryReportsOtherScanErrors(t *testing.T) {
+	for _, origin := range []string{"Info", "WalkDir"} {
+		t.Run(origin, func(t *testing.T) {
+			directory := t.TempDir()
+			filePath := filepath.Join(directory, "recording.mp4")
+			if err := os.WriteFile(filePath, []byte("video"), 0o644); err != nil {
+				t.Fatal(err)
+			}
+			entries, err := os.ReadDir(directory)
+			if err != nil {
+				t.Fatal(err)
+			}
+			wantErr := &fs.PathError{Op: "lstat", Path: filePath, Err: fs.ErrPermission}
+			files, err := readRecordingDirectory(directory, func(root string, visit fs.WalkDirFunc) error {
+				if origin == "Info" {
+					return visit(filePath, recordingDirEntryInfoError{DirEntry: entries[0], err: wantErr}, nil)
+				}
+				return visit(filePath, nil, wantErr)
+			})
+			if !errors.Is(err, fs.ErrPermission) || files != nil {
+				t.Fatalf("scan = %v, %v; want permission error without partial results", files, err)
+			}
+		})
+	}
+}
+
 func TestDecryptDirectoryIncludesDatedRecordingsAndSkipsOutputs(t *testing.T) {
 	directory := t.TempDir()
 	key := []byte("0123456789abcdef0123456789abcdef")