Fix pending-upload download race that purges local PVE files

debian/changelog

@@ -16,6 +16,22 @@ proxs3 (0.7.0-1) unstable; urgency=medium
     multi-GB backups loop forever (download -> evict -> retry). Files
     modified within the last 30 minutes are now exempt from size-based
     eviction; a warning is logged when the cache stays over budget.
+  * Fix /v1/download destroying locally created files not yet uploaded to
+    S3: a HeadObject 404 was treated as "deleted on S3" and purged the
+    cached file even when it had never been uploaded (e.g. cloud-init
+    user-data ISOs written by PVE and attached immediately, inside the
+    watcher's debounce window), losing the file permanently. Files are now
+    explicitly tracked as pending upload: the watcher writes a durable
+    pending-upload marker when it detects a local write, and while a file
+    is pending (or has no metadata at all) download serves the local copy
+    and queues the upload instead of purging it or overwriting it with a
+    stale S3 version of the same key. Files confirmed synced still fail
+    closed on 404. Pending files survive restarts (the watcher scans the
+    cache for unconfirmed files on startup) and are exempt from cache
+    eviction until uploaded. An upload trigger arriving mid-transfer now
+    reruns the upload instead of being dropped, delete aborts any in-flight
+    upload of the same file so it can't resurrect the object, and uploads
+    record the S3 ETag so the next download doesn't needlessly re-fetch.
   * Fix PVE restore "connection error timeout" in the UI: path() no longer
     triggers a synchronous download from the API thread for content types
     where activate_volume is called (backup, iso, vztmpl, images). PVE's

internal/api/api.go

@@ -30,8 +30,28 @@ type Server struct {
 	usage    map[string]int64
 	healthMu sync.RWMutex
 	clientMu sync.RWMutex
-	listener net.Listener
-	server   *http.Server
+	// uploads tracks in-flight uploads by local path. A second trigger while
+	// an upload runs marks it for a rerun instead of being dropped, and
+	// handleDelete can cancel the in-flight transfer.
+	uploads   map[string]*uploadState
+	uploadsMu sync.Mutex
+	// uploadRequests feeds paths into the watcher's debounce queue, so
+	// uploads triggered outside fsnotify (e.g. by handleDownload) still get
+	// the stability window and in-use check before transferring.
+	uploadRequests chan string
+	// selfWrites records paths the daemon itself is about to create inside
+	// the watched cache dirs (download renames, upload links), so the watcher
+	// doesn't mistake them for new PVE writes and mark them pending upload.
+	selfWrites sync.Map
+	listener   net.Listener
+	server     *http.Server
+}
+
+// uploadState tracks one in-flight upload for dedup, rerun, and cancellation.
+type uploadState struct {
+	rerun    bool
+	canceled bool
+	cancel   context.CancelFunc
 }
 
 // New creates a new API server.
@@ -42,11 +62,13 @@ func New(cfg *config.DaemonConfig) (*Server, error) {
 	}
 
 	s := &Server{
-		cfg:     cfg,
-		clients: make(map[string]s3client.S3Client),
-		cache:   fc,
-		health:  make(map[string]bool),
-		usage:   make(map[string]int64),
+		cfg:            cfg,
+		clients:        make(map[string]s3client.S3Client),
+		cache:          fc,
+		health:         make(map[string]bool),
+		usage:          make(map[string]int64),
+		uploads:        make(map[string]*uploadState),
+		uploadRequests: make(chan string, 1024),
 	}
 
 	if err := s.initClients(cfg); err != nil {
@@ -359,6 +381,24 @@ func (s *Server) handleDownload(w http.ResponseWriter, r *http.Request) {
 	headCancel()
 
 	if cached := s.cache.Path(storageID, key); cached != "" {
+		// Snapshot metadata before branching on the HEAD result: the watcher
+		// may finish an upload (flipping pending → synced) between the HEAD
+		// call and this read, and reading the metadata second keeps that race
+		// on the safe side (a pending verdict serves the local file).
+		meta := s.cache.GetMeta(storageID, key)
+		if meta == nil || meta.PendingUpload {
+			// The local file is authoritative: it was created or rewritten by
+			// PVE and the watcher hasn't uploaded it yet. Metadata is only
+			// written after a confirmed S3 operation, so missing metadata
+			// means the same thing. Whatever HEAD said — 404 because the
+			// object isn't on S3 yet, or 200 for a stale previous version
+			// under the same key — the local copy wins: purging or
+			// re-downloading here would destroy data that never reached S3.
+			log.Printf("download: %s/%s pending upload, serving local copy", storageID, key)
+			s.requestUpload(cached)
+			writeJSON(w, map[string]string{"path": cached})
+			return
+		}
 		if errors.Is(headErr, s3client.ErrNotFound) {
 			// Object deleted on S3 — purge stale cache and fail closed.
 			// Use Remove (not Invalidate) so the immutable flag PVE sets on
@@ -439,7 +479,11 @@ func (s *Server) handleDownload(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
+	// The rename lands inside a watched dir; tell the watcher it's ours so it
+	// doesn't mark the freshly downloaded file as a pending PVE write.
+	s.selfWrites.Store(finalPath, time.Now())
 	if err := os.Rename(tmpPath, finalPath); err != nil {
+		s.selfWrites.Delete(finalPath)
 		os.Remove(tmpPath)
 		log.Printf("download: rename %s -> %s failed: %v", tmpPath, finalPath, err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
@@ -494,11 +538,20 @@ func (s *Server) handleUpload(w http.ResponseWriter, r *http.Request) {
 	log.Printf("upload: uploaded %s to s3://%s (%.1f MB, %s)",
 		key, storageID, float64(info.Size())/(1024*1024), time.Since(start).Round(time.Millisecond))
 
-	// Cache the uploaded file with current metadata
+	// Cache the uploaded file with current metadata. Record the ETag S3
+	// assigned so the staleness check recognizes the cached copy as current
+	// instead of re-downloading it on first access.
 	uploadMeta := cache.FileMeta{
 		Size:         info.Size(),
 		LastModified: time.Now(),
 	}
+	if head, err := client.HeadObject(r.Context(), key); err == nil {
+		uploadMeta.ETag = head.ETag
+		uploadMeta.LastModified = head.LastModified
+	}
+	// Link may create the file inside a watched dir; tell the watcher it's
+	// ours so it doesn't mark the just-uploaded file as a pending PVE write.
+	s.selfWrites.Store(s.cache.ExpectedPath(storageID, key), time.Now())
 	s.cache.Link(storageID, key, localPath, uploadMeta)
 
 	writeJSON(w, map[string]string{"status": "ok"})
@@ -516,6 +569,10 @@ func (s *Server) handleDelete(w http.ResponseWriter, r *http.Request) {
 
 	log.Printf("delete: deleting %s from s3://%s", key, storageID)
 
+	// Abort any in-flight background upload of this file first — a PutObject
+	// completing after the DeleteObject would resurrect the object on S3.
+	s.cancelUpload(s.cache.ExpectedPath(storageID, key))
+
 	if err := client.DeleteObject(r.Context(), key); err != nil {
 		log.Printf("delete: failed to delete %s from %s: %v", key, storageID, err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)

internal/api/handler_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -20,13 +21,18 @@ import (
 // mockS3Client implements s3client.S3Client for testing.
 type mockS3Client struct {
 	id      string
+	mu      sync.Mutex            // guards objects for tests with concurrent uploads
 	objects map[string]mockObject // key -> object
 	healthy bool
 	listErr error
 	headErr error
 	getErr  error
 	putErr  error
 	delErr  error
+	// Optional hooks for concurrency tests: PutObject signals putStarted (if
+	// set), then blocks until putRelease is closed or the context is canceled.
+	putStarted chan string
+	putRelease chan struct{}
 }
 
 type mockObject struct {
@@ -53,10 +59,21 @@ func (m *mockS3Client) HeadBucket(ctx context.Context) error {
 	return nil
 }
 
+// object returns a stored object under the lock, for assertions in tests
+// that run concurrent uploads.
+func (m *mockS3Client) object(key string) (mockObject, bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	obj, ok := m.objects[key]
+	return obj, ok
+}
+
 func (m *mockS3Client) ListObjects(ctx context.Context, prefix string) ([]s3client.ObjectInfo, error) {
 	if m.listErr != nil {
 		return nil, m.listErr
 	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	var objects []s3client.ObjectInfo
 	for key, obj := range m.objects {
 		if strings.HasPrefix(key, prefix) {
@@ -75,7 +92,7 @@ func (m *mockS3Client) HeadObject(ctx context.Context, key string) (*s3client.Ob
 	if m.headErr != nil {
 		return nil, m.headErr
 	}
-	obj, ok := m.objects[key]
+	obj, ok := m.object(key)
 	if !ok {
 		return nil, fmt.Errorf("not found: %s", key)
 	}
@@ -91,7 +108,7 @@ func (m *mockS3Client) GetObject(ctx context.Context, key string) (*s3client.Get
 	if m.getErr != nil {
 		return nil, m.getErr
 	}
-	obj, ok := m.objects[key]
+	obj, ok := m.object(key)
 	if !ok {
 		return nil, fmt.Errorf("not found: %s", key)
 	}
@@ -107,7 +124,7 @@ func (m *mockS3Client) DownloadToFile(ctx context.Context, key string, w io.Writ
 	if m.getErr != nil {
 		return 0, m.getErr
 	}
-	obj, ok := m.objects[key]
+	obj, ok := m.object(key)
 	if !ok {
 		return 0, fmt.Errorf("not found: %s", key)
 	}
@@ -119,7 +136,19 @@ func (m *mockS3Client) PutObject(ctx context.Context, key string, body io.Reader
 	if m.putErr != nil {
 		return m.putErr
 	}
+	if m.putStarted != nil {
+		m.putStarted <- key
+	}
+	if m.putRelease != nil {
+		select {
+		case <-m.putRelease:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
 	data, _ := io.ReadAll(body)
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	m.objects[key] = mockObject{
 		data:         string(data),
 		size:         size,
@@ -133,11 +162,15 @@ func (m *mockS3Client) DeleteObject(ctx context.Context, key string) error {
 	if m.delErr != nil {
 		return m.delErr
 	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	delete(m.objects, key)
 	return nil
 }
 
 func (m *mockS3Client) CopyObject(ctx context.Context, srcKey, dstKey string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	obj, ok := m.objects[srcKey]
 	if !ok {
 		return fmt.Errorf("source key %s not found", srcKey)
@@ -172,9 +205,11 @@ func newTestServer(t *testing.T, mock *mockS3Client) *Server {
 		clients: map[string]s3client.S3Client{
 			mock.id: mock,
 		},
-		cache:  fc,
-		health: map[string]bool{mock.id: mock.healthy},
-		usage:  map[string]int64{mock.id: 0},
+		cache:          fc,
+		health:         map[string]bool{mock.id: mock.healthy},
+		usage:          map[string]int64{mock.id: 0},
+		uploads:        make(map[string]*uploadState),
+		uploadRequests: make(chan string, 1024),
 	}
 	return s
 }
@@ -646,6 +681,108 @@ func TestHandleDownload_ObjectDeletedOnS3_RemovesCache(t *testing.T) {
 	}
 }
 
+func TestHandleDownload_LocalPendingUpload_ServedNotPurged(t *testing.T) {
+	mock := newMockClient("s3test")
+	// The object does not exist on S3 yet — the watcher hasn't uploaded it.
+	mock.headErr = s3client.ErrNotFound
+	s := newTestServer(t, mock)
+
+	// PVE writes files directly into the cache dir; no meta sidecar exists yet.
+	key := "template/iso/user-data-200.iso"
+	localPath := filepath.Join(s.cfg.CacheDir, "s3test", "template", "iso", "user-data-200.iso")
+	os.MkdirAll(filepath.Dir(localPath), 0755)
+	os.WriteFile(localPath, []byte("cloud-init payload"), 0644)
+
+	req := httptest.NewRequest("GET", "/v1/download?storage=s3test&key="+key, nil)
+	w := httptest.NewRecorder()
+	s.handleDownload(w, req)
+
+	if w.Code != 200 {
+		t.Fatalf("expected 200 for local file pending upload, got %d: %s", w.Code, w.Body.String())
+	}
+	var resp map[string]string
+	json.NewDecoder(w.Body).Decode(&resp)
+	if resp["path"] != localPath {
+		t.Errorf("expected local path %q, got %q", localPath, resp["path"])
+	}
+	if _, err := os.Stat(localPath); err != nil {
+		t.Errorf("expected local file to survive, stat failed: %v", err)
+	}
+
+	// The file must be queued for upload through the watcher's debounce loop.
+	select {
+	case p := <-s.uploadRequests:
+		if p != localPath {
+			t.Errorf("expected upload request for %q, got %q", localPath, p)
+		}
+	default:
+		t.Error("expected the served pending file to be queued for upload")
+	}
+}
+
+func TestHandleDownload_PendingRewrite_ServesLocalNotS3(t *testing.T) {
+	mock := newMockClient("s3test")
+	key := "template/iso/user-data-118.iso"
+	// An older version of the object still exists on S3 under the same key.
+	mock.objects[key] = mockObject{data: "old", size: 3, etag: "\"old\"", lastModified: time.Now().Add(-time.Hour)}
+	s := newTestServer(t, mock)
+
+	// PVE rewrote the file locally; the watcher marked it pending upload.
+	// The stale S3 copy must not clobber it via the staleness check.
+	localPath := filepath.Join(s.cfg.CacheDir, "s3test", "template", "iso", "user-data-118.iso")
+	os.MkdirAll(filepath.Dir(localPath), 0755)
+	os.WriteFile(localPath, []byte("new payload"), 0644)
+	s.cache.StoreMeta("s3test", key, cache.FileMeta{Size: 11, LastModified: time.Now(), PendingUpload: true})
+
+	req := httptest.NewRequest("GET", "/v1/download?storage=s3test&key="+key, nil)
+	w := httptest.NewRecorder()
+	s.handleDownload(w, req)
+
+	if w.Code != 200 {
+		t.Fatalf("expected 200 for pending rewrite, got %d: %s", w.Code, w.Body.String())
+	}
+	var resp map[string]string
+	json.NewDecoder(w.Body).Decode(&resp)
+	if resp["path"] != localPath {
+		t.Errorf("expected local path %q, got %q", localPath, resp["path"])
+	}
+	data, err := os.ReadFile(localPath)
+	if err != nil || string(data) != "new payload" {
+		t.Errorf("expected local rewrite to survive, got %q (err %v)", data, err)
+	}
+	select {
+	case <-s.uploadRequests:
+	default:
+		t.Error("expected the pending rewrite to be queued for upload")
+	}
+}
+
+func TestHandleDownload_ObjectDeletedOnS3_WatcherUploadedMeta_RemovesCache(t *testing.T) {
+	mock := newMockClient("s3test")
+	mock.headErr = s3client.ErrNotFound
+	s := newTestServer(t, mock)
+
+	// Simulate a locally created file the watcher already uploaded: content on
+	// disk plus the meta the watcher writes after PutObject (size, no ETag).
+	// With confirmed S3 provenance, a 404 means deleted on S3 → fail closed.
+	key := "template/iso/user-data-118.iso"
+	localPath := filepath.Join(s.cfg.CacheDir, "s3test", "template", "iso", "user-data-118.iso")
+	os.MkdirAll(filepath.Dir(localPath), 0755)
+	os.WriteFile(localPath, []byte("stale"), 0644)
+	s.cache.StoreMeta("s3test", key, cache.FileMeta{Size: 5, LastModified: time.Now()})
+
+	req := httptest.NewRequest("GET", "/v1/download?storage=s3test&key="+key, nil)
+	w := httptest.NewRecorder()
+	s.handleDownload(w, req)
+
+	if w.Code != 404 {
+		t.Fatalf("expected 404 when uploaded object was deleted on S3, got %d: %s", w.Code, w.Body.String())
+	}
+	if p := s.cache.Path("s3test", key); p != "" {
+		t.Errorf("expected cache purged, but Path returned %q", p)
+	}
+}
+
 func TestHandleDownload_S3Unreachable_NoCache(t *testing.T) {
 	mock := newMockClient("s3test")
 	mock.getErr = fmt.Errorf("connection refused")