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

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

README.md

@@ -464,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/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 })