diff --git a/debian/changelog b/debian/changelog index 8d5da42..4d5ce1f 100644 --- a/debian/changelog +++ b/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 diff --git a/internal/api/api.go b/internal/api/api.go index b4ecf95..4c37891 100644 --- a/internal/api/api.go +++ b/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) diff --git a/internal/api/handler_test.go b/internal/api/handler_test.go index 9cba7c5..4bde95a 100644 --- a/internal/api/handler_test.go +++ b/internal/api/handler_test.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -20,6 +21,7 @@ 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 @@ -27,6 +29,10 @@ type mockS3Client struct { 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") diff --git a/internal/api/resync.go b/internal/api/resync.go index 4722a52..12ec5a8 100644 --- a/internal/api/resync.go +++ b/internal/api/resync.go @@ -152,10 +152,19 @@ func (s *Server) handleResync(w http.ResponseWriter, r *http.Request) { return nil } - s.cache.StoreMeta(storageID, s3Key, cache.FileMeta{ + // Record the ETag S3 assigned so the staleness check recognizes the + // cached copy as current instead of re-downloading it on first access. + syncedMeta := cache.FileMeta{ Size: size, LastModified: time.Now(), - }) + } + hCtx, hCancel := context.WithTimeout(r.Context(), 10*time.Second) + if head, err := client.HeadObject(hCtx, s3Key); err == nil { + syncedMeta.ETag = head.ETag + syncedMeta.LastModified = head.LastModified + } + hCancel() + s.cache.StoreMeta(storageID, s3Key, syncedMeta) uploaded++ return nil }) diff --git a/internal/api/watcher.go b/internal/api/watcher.go index b86e456..7747ba3 100644 --- a/internal/api/watcher.go +++ b/internal/api/watcher.go @@ -41,14 +41,37 @@ func (s *Server) watchCacheDirs() { ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() + // Files written while the daemon wasn't running have no fsnotify + // event; queue anything on disk that isn't confirmed uploaded so + // pending uploads survive restarts. + s.scanPendingUploads(pending) + for { select { case event, ok := <-watcher.Events: if !ok { return } - if event.Op&(fsnotify.Create|fsnotify.Write) != 0 { - pending[event.Name] = time.Now() + if event.Op&(fsnotify.Create|fsnotify.Write) == 0 { + continue + } + if s.consumeSelfWrite(event.Name) { + // Our own download rename or upload link — already synced. + continue + } + if _, seen := pending[event.Name]; !seen { + // New pending episode: durably mark the local file as + // authoritative before anything can act on the old state. + s.markPendingUpload(event.Name) + } + pending[event.Name] = time.Now() + + case path := <-s.uploadRequests: + // Upload requested outside fsnotify (e.g. handleDownload + // served a pending file). Runs through the same debounce + // and in-use checks as watcher-detected writes. + if _, seen := pending[path]; !seen { + pending[path] = time.Now() } case err, ok := <-watcher.Errors: @@ -160,31 +183,159 @@ func fileInUse(path string) bool { return false } -// uploadNewFile detects the storage ID and S3 key from a local cache path -// and uploads the file to S3. +// cacheRelKey maps a path inside the cache dir to its storage ID and S3 key. +// Parses: /var/cache/proxs3/// +func (s *Server) cacheRelKey(localPath string) (storageID, s3Key string, ok bool) { + rel, err := filepath.Rel(s.cfg.CacheDir, localPath) + if err != nil { + return "", "", false + } + // rel is like "s3test/template/iso/debian.iso" + parts := strings.SplitN(rel, string(os.PathSeparator), 2) + if len(parts) != 2 { + return "", "", false + } + // Normalize path separators to forward slashes for S3 + return parts[0], filepath.ToSlash(parts[1]), true +} + +// consumeSelfWrite reports whether the daemon itself just created this path +// (download rename, upload link). Entries expire so a stale one can't +// swallow a later genuine PVE write event for the same path. +func (s *Server) consumeSelfWrite(path string) bool { + v, ok := s.selfWrites.LoadAndDelete(path) + if !ok { + return false + } + return time.Since(v.(time.Time)) < 10*time.Second +} + +// markPendingUpload durably marks a file PVE just wrote into the cache dir, +// so the local copy is treated as authoritative — by handleDownload, by +// eviction, and across daemon restarts — until the upload completes. +func (s *Server) markPendingUpload(localPath string) { + if strings.HasSuffix(localPath, ".tmp") || strings.HasSuffix(localPath, ".meta") { + return + } + info, err := os.Stat(localPath) + if err != nil || info.IsDir() { + return + } + storageID, s3Key, ok := s.cacheRelKey(localPath) + if !ok { + return + } + s.cache.StoreMeta(storageID, s3Key, cache.FileMeta{ + Size: info.Size(), + LastModified: info.ModTime(), + PendingUpload: true, + }) +} + +// scanPendingUploads walks the storage cache trees and queues files that are +// not confirmed uploaded to S3: no metadata (written locally, never +// processed) or an explicit pending-upload marker. fsnotify alone can't see +// files written while the daemon wasn't running. +func (s *Server) scanPendingUploads(pending map[string]time.Time) { + s.clientMu.RLock() + storages := make([]string, 0, len(s.clients)) + for id := range s.clients { + storages = append(storages, id) + } + s.clientMu.RUnlock() + + for _, storageID := range storages { + baseDir := filepath.Join(s.cfg.CacheDir, storageID) + filepath.Walk(baseDir, func(path string, info os.FileInfo, err error) error { + if err != nil || info.IsDir() { + return nil + } + if strings.HasSuffix(path, ".tmp") || strings.HasSuffix(path, ".meta") { + return nil + } + rel, err := filepath.Rel(baseDir, path) + if err != nil { + return nil + } + s3Key := filepath.ToSlash(rel) + if meta := s.cache.GetMeta(storageID, s3Key); meta == nil || meta.PendingUpload { + log.Printf("watcher: %s/%s not confirmed on S3, queueing upload", storageID, s3Key) + pending[path] = time.Now() + } + return nil + }) + } +} + +// requestUpload queues a file for upload via the watcher's debounce loop, so +// it gets the same stability window and in-use check as fsnotify-detected +// writes. Non-blocking: if the queue is full (or the watcher isn't running), +// the file is still safe — it stays marked pending and the restart scan or +// the next write event picks it up. +func (s *Server) requestUpload(localPath string) { + select { + case s.uploadRequests <- localPath: + default: + log.Printf("watcher: upload queue full, dropping request for %s", localPath) + } +} + +// cancelUpload aborts any in-flight background upload for localPath and +// suppresses a queued rerun. Used by delete so a PutObject completing after +// the DeleteObject can't resurrect a just-deleted object. +func (s *Server) cancelUpload(localPath string) { + s.uploadsMu.Lock() + defer s.uploadsMu.Unlock() + if st := s.uploads[localPath]; st != nil { + st.canceled = true + st.rerun = false + if st.cancel != nil { + st.cancel() + } + } +} + +// uploadNewFile uploads a local cache file to S3, deduplicating concurrent +// triggers. A trigger that arrives while an upload is already running (e.g. +// the file was rewritten mid-transfer) schedules a rerun after it finishes +// instead of being dropped. func (s *Server) uploadNewFile(localPath string) { // Skip .meta sidecar files — these are cache metadata, not real content if strings.HasSuffix(localPath, ".meta") { return } - // Parse: /var/cache/proxs3/// - rel, err := filepath.Rel(s.cfg.CacheDir, localPath) - if err != nil { - log.Printf("watcher: can't determine relative path for %s: %v", localPath, err) + s.uploadsMu.Lock() + if st := s.uploads[localPath]; st != nil { + st.rerun = true + s.uploadsMu.Unlock() return } + st := &uploadState{} + s.uploads[localPath] = st + s.uploadsMu.Unlock() - // rel is like "s3test/template/iso/debian.iso" - parts := strings.SplitN(rel, string(os.PathSeparator), 2) - if len(parts) != 2 { + for { + s.uploadOnce(localPath, st) + + s.uploadsMu.Lock() + if st.rerun && !st.canceled { + st.rerun = false + s.uploadsMu.Unlock() + continue + } + delete(s.uploads, localPath) + s.uploadsMu.Unlock() return } - storageID := parts[0] - s3Key := parts[1] +} - // Normalize path separators to forward slashes for S3 - s3Key = filepath.ToSlash(s3Key) +// uploadOnce performs a single upload attempt for uploadNewFile. +func (s *Server) uploadOnce(localPath string, st *uploadState) { + storageID, s3Key, ok := s.cacheRelKey(localPath) + if !ok { + return + } client, ok := s.getClient(storageID) if !ok { @@ -206,9 +357,9 @@ func (s *Server) uploadNewFile(localPath string) { } // Skip if cache metadata shows this file is already in sync with S3. - // Files written by handleDownload (from S3) or handleUpload (already pushed) - // have metadata with matching size — no need to re-upload. - if meta := s.cache.GetMeta(storageID, s3Key); meta != nil && meta.Size == info.Size() { + // Files written by handleDownload (from S3) or handleUpload (already + // pushed) have non-pending metadata with matching size. + if meta := s.cache.GetMeta(storageID, s3Key); meta != nil && !meta.PendingUpload && meta.Size == info.Size() { log.Printf("watcher: skipping %s in %s (already synced to S3)", s3Key, storageID) return } @@ -218,17 +369,38 @@ func (s *Server) uploadNewFile(localPath string) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() + s.uploadsMu.Lock() + if st.canceled { + s.uploadsMu.Unlock() + return + } + st.cancel = cancel + s.uploadsMu.Unlock() if err := client.PutObject(ctx, s3Key, f, info.Size()); err != nil { log.Printf("watcher: upload failed for %s: %v", s3Key, err) return } - // Update cache metadata + s.uploadsMu.Lock() + canceled := st.canceled + s.uploadsMu.Unlock() + if canceled { + // Deleted while the last bytes were in flight — don't write metadata + // for a file delete is about to remove. + return + } + + // Update cache metadata; record the ETag S3 assigned so the staleness + // check recognizes the cached copy as current instead of re-downloading. meta := cache.FileMeta{ Size: info.Size(), LastModified: time.Now(), } + if head, err := client.HeadObject(ctx, s3Key); err == nil { + meta.ETag = head.ETag + meta.LastModified = head.LastModified + } s.cache.StoreMeta(storageID, s3Key, meta) log.Printf("watcher: uploaded %s to %s successfully", s3Key, storageID) diff --git a/internal/api/watcher_test.go b/internal/api/watcher_test.go index 6db22cd..f942d75 100644 --- a/internal/api/watcher_test.go +++ b/internal/api/watcher_test.go @@ -1,10 +1,14 @@ package api import ( + "net/http/httptest" "os" "path/filepath" "strings" "testing" + "time" + + "github.com/sol1/proxs3/internal/cache" ) func TestFileInUse_NonexistentFile(t *testing.T) { @@ -110,3 +114,151 @@ func TestUploadNewFile_PathParsing(t *testing.T) { } } } + +func TestMarkPendingUpload(t *testing.T) { + s := newTestServer(t, newMockClient("s3test")) + dir := filepath.Join(s.cfg.CacheDir, "s3test", "snippets") + os.MkdirAll(dir, 0755) + + path := filepath.Join(dir, "new.yaml") + os.WriteFile(path, []byte("hello"), 0644) + s.markPendingUpload(path) + + meta := s.cache.GetMeta("s3test", "snippets/new.yaml") + if meta == nil || !meta.PendingUpload { + t.Fatalf("expected pending-upload marker, got %+v", meta) + } + if meta.Size != 5 { + t.Errorf("expected size 5 in marker, got %d", meta.Size) + } + + // Directories and .tmp files must not get markers. + s.markPendingUpload(dir) + tmpPath := filepath.Join(dir, "partial.yaml.tmp") + os.WriteFile(tmpPath, []byte("x"), 0644) + s.markPendingUpload(tmpPath) + if meta := s.cache.GetMeta("s3test", "snippets"); meta != nil { + t.Error("expected no marker for a directory") + } + if meta := s.cache.GetMeta("s3test", "snippets/partial.yaml.tmp"); meta != nil { + t.Error("expected no marker for a .tmp file") + } +} + +func TestScanPendingUploads(t *testing.T) { + s := newTestServer(t, newMockClient("s3test")) + dir := filepath.Join(s.cfg.CacheDir, "s3test", "snippets") + os.MkdirAll(dir, 0755) + + // No metadata: written locally, never processed - must be queued. + orphan := filepath.Join(dir, "orphan.yaml") + os.WriteFile(orphan, []byte("a"), 0644) + // Explicit pending marker - must be queued. + marked := filepath.Join(dir, "marked.yaml") + os.WriteFile(marked, []byte("bb"), 0644) + s.cache.StoreMeta("s3test", "snippets/marked.yaml", cache.FileMeta{Size: 2, LastModified: time.Now(), PendingUpload: true}) + // Confirmed synced - must not be queued. + synced := filepath.Join(dir, "synced.yaml") + os.WriteFile(synced, []byte("ccc"), 0644) + s.cache.StoreMeta("s3test", "snippets/synced.yaml", cache.FileMeta{Size: 3, LastModified: time.Now(), ETag: "\"e\""}) + // In-progress temp file - must not be queued. + os.WriteFile(filepath.Join(dir, "partial.yaml.tmp"), []byte("d"), 0644) + + pending := make(map[string]time.Time) + s.scanPendingUploads(pending) + + if _, ok := pending[orphan]; !ok { + t.Error("expected metadata-less file to be queued") + } + if _, ok := pending[marked]; !ok { + t.Error("expected pending-marked file to be queued") + } + if len(pending) != 2 { + t.Errorf("expected exactly 2 queued files, got %d: %v", len(pending), pending) + } +} + +func TestUploadNewFile_RerunAfterTriggerDuringUpload(t *testing.T) { + mock := newMockClient("s3test") + mock.putStarted = make(chan string, 2) + mock.putRelease = make(chan struct{}) + s := newTestServer(t, mock) + + key := "snippets/user.yaml" + localPath := filepath.Join(s.cfg.CacheDir, "s3test", "snippets", "user.yaml") + os.MkdirAll(filepath.Dir(localPath), 0755) + os.WriteFile(localPath, []byte("v1"), 0644) + + done := make(chan struct{}) + go func() { + s.uploadNewFile(localPath) + close(done) + }() + <-mock.putStarted // first upload is in flight + + // The file is rewritten mid-upload; this trigger must not be dropped. + os.WriteFile(localPath, []byte("v2-longer"), 0644) + s.uploadNewFile(localPath) // returns immediately, schedules a rerun + + close(mock.putRelease) + select { + case <-mock.putStarted: // rerun started + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the rerun upload to start") + } + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for uploads to finish") + } + + obj, ok := mock.object(key) + if !ok { + t.Fatal("expected object on S3 after rerun") + } + if obj.data != "v2-longer" { + t.Errorf("expected rerun to upload rewritten content, got %q", obj.data) + } +} + +func TestHandleDelete_CancelsInFlightUpload(t *testing.T) { + mock := newMockClient("s3test") + mock.putStarted = make(chan string, 1) + mock.putRelease = make(chan struct{}) // never released: only ctx cancel unblocks + s := newTestServer(t, mock) + + key := "snippets/user.yaml" + localPath := filepath.Join(s.cfg.CacheDir, "s3test", "snippets", "user.yaml") + os.MkdirAll(filepath.Dir(localPath), 0755) + os.WriteFile(localPath, []byte("payload"), 0644) + + done := make(chan struct{}) + go func() { + s.uploadNewFile(localPath) + close(done) + }() + <-mock.putStarted // upload is in flight + + req := httptest.NewRequest("DELETE", "/v1/delete?storage=s3test&key="+key, nil) + w := httptest.NewRecorder() + s.handleDelete(w, req) + + if w.Code != 200 { + t.Fatalf("expected 200 from delete, got %d: %s", w.Code, w.Body.String()) + } + select { + case <-done: // upload aborted via context cancel + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for canceled upload to return") + } + + if _, ok := mock.object(key); ok { + t.Error("expected no object on S3 after delete canceled the upload") + } + if p := s.cache.Path("s3test", key); p != "" { + t.Errorf("expected local file removed by delete, got %q", p) + } + if meta := s.cache.GetMeta("s3test", key); meta != nil { + t.Errorf("expected no metadata after delete, got %+v", meta) + } +} diff --git a/internal/cache/cache.go b/internal/cache/cache.go index fa42bb1..63e245f 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -30,6 +30,12 @@ type FileMeta struct { LastModified time.Time `json:"last_modified"` Size int64 `json:"size"` CachedAt time.Time `json:"cached_at"` + // PendingUpload marks a file whose local content is authoritative: it was + // created or rewritten locally (by PVE) and has not been uploaded to S3 + // yet. Such files must never be purged or overwritten from S3. The zero + // value means "in sync with S3", so metadata written by older versions + // keeps its meaning. + PendingUpload bool `json:"pending_upload,omitempty"` } // FileCache provides a local filesystem cache for S3 objects. @@ -282,6 +288,17 @@ func (fc *FileCache) EvictByAge(storageID string, maxAge time.Duration) int { return nil } if info.ModTime().Before(cutoff) { + // Never evict a file that hasn't been uploaded to S3 — the local + // copy is the only copy. No metadata means the same thing: the + // file was written locally and not yet processed. Legacy .meta + // sidecars are cleaned up alongside their content file as before. + if !strings.HasSuffix(path, ".meta") { + if rel, err := filepath.Rel(storageDir, path); err == nil { + if meta := fc.GetMeta(storageID, filepath.ToSlash(rel)); meta == nil || meta.PendingUpload { + return nil + } + } + } fc.mu.Lock() err := os.Remove(path) if err != nil && !os.IsNotExist(err) { @@ -346,6 +363,18 @@ func (fc *FileCache) evictIfNeeded() { protectedSize += info.Size() return nil } + // Never size-evict a file that hasn't been uploaded to S3 — the + // local copy is the only copy. Partial downloads under .downloads/ + // stay evictable (they never have metadata by design). + if rel, err := filepath.Rel(fc.baseDir, path); err == nil { + parts := strings.SplitN(filepath.ToSlash(rel), "/", 2) + if len(parts) == 2 && parts[0] != ".downloads" { + if meta := fc.GetMeta(parts[0], parts[1]); meta == nil || meta.PendingUpload { + protectedSize += info.Size() + return nil + } + } + } files = append(files, cachedFile{ path: path, size: info.Size(), diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 461b21f..34e6811 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -731,3 +731,66 @@ func TestMultipleStorages(t *testing.T) { t.Errorf("unexpected content for storage-b: %s", contentB) } } + +func TestEvictByAge_SkipsPendingUpload(t *testing.T) { + dir := t.TempDir() + fc, err := New(dir, 100) + if err != nil { + t.Fatalf("New failed: %v", err) + } + + tenDaysAgo := time.Now().Add(-10 * 24 * time.Hour) + + // Old file with a pending-upload marker - the only copy, must survive. + fc.Store("s", "iso/pending.iso", strings.NewReader("aaaaa"), FileMeta{Size: 5, PendingUpload: true}) + os.Chtimes(fc.path("s", "iso/pending.iso"), tenDaysAgo, tenDaysAgo) + + // Old file with no metadata - written locally, never processed; must survive. + orphan := fc.path("s", "iso/orphan.iso") + os.MkdirAll(filepath.Dir(orphan), 0755) + os.WriteFile(orphan, []byte("bbbbb"), 0644) + os.Chtimes(orphan, tenDaysAgo, tenDaysAgo) + + // Old file confirmed on S3 - evictable. + fc.Store("s", "iso/synced.iso", strings.NewReader("ccccc"), FileMeta{Size: 5, ETag: "\"e\""}) + os.Chtimes(fc.path("s", "iso/synced.iso"), tenDaysAgo, tenDaysAgo) + + removed := fc.EvictByAge("s", 7*24*time.Hour) + if removed != 1 { + t.Errorf("expected 1 file removed, got %d", removed) + } + if !fc.Has("s", "iso/pending.iso") { + t.Error("pending-upload file must not be age-evicted") + } + if !fc.Has("s", "iso/orphan.iso") { + t.Error("metadata-less file must not be age-evicted") + } + if fc.Has("s", "iso/synced.iso") { + t.Error("expected synced file to be age-evicted") + } +} + +func TestEviction_SkipsPendingUpload(t *testing.T) { + dir := t.TempDir() + fc, err := New(dir, 1) // 1 MB cap + if err != nil { + t.Fatalf("New failed: %v", err) + } + + old := time.Now().Add(-2 * time.Hour) // well past evictMinAge + + fc.Store("s", "dump/pending.vma", bytes.NewReader(make([]byte, 800*1024)), FileMeta{Size: 800 * 1024, PendingUpload: true}) + os.Chtimes(fc.path("s", "dump/pending.vma"), old, old) + + fc.Store("s", "dump/synced.vma", bytes.NewReader(make([]byte, 800*1024)), FileMeta{Size: 800 * 1024, ETag: "\"e\""}) + os.Chtimes(fc.path("s", "dump/synced.vma"), old, old) + + fc.evictIfNeeded() + + if !fc.Has("s", "dump/pending.vma") { + t.Error("pending-upload file must not be size-evicted") + } + if fc.Has("s", "dump/synced.vma") { + t.Error("expected synced file to be size-evicted") + } +}