feat: the backup can include the music, sending each library file once to music/ and again only when it changes (#6774)
release / govulncheck (push) Successful in 15s
release / web (push) Failing after 1m17s
release / go (push) Successful in 1m33s
release / integration (push) Successful in 5m16s
release / android (push) Successful in 5m56s
release / Build signed APK (push) Successful in 6m13s
release / Attach APK to the Release (tag releases only) (push) Skipped
release / Build + push container image (push) Skipped
release / Verify release artifacts (tag releases only) (push) Skipped
release / govulncheck (push) Successful in 15s
release / web (push) Failing after 1m17s
release / go (push) Successful in 1m33s
release / integration (push) Successful in 5m16s
release / android (push) Successful in 5m56s
release / Build signed APK (push) Successful in 6m13s
release / Attach APK to the Release (tag releases only) (push) Skipped
release / Build + push container image (push) Skipped
release / Verify release artifacts (tag releases only) (push) Skipped
With "Include the music" on, each night's run sends the library files the destination doesn't hold yet, after the database file has landed, for up to six hours; a large library spans several nights the first time. Files keep their path under the library folders' common root, beside database/ in the same folder or bucket. backup_music_files records each file's size and mtime as sent, so a night sends only new and changed files, and a rescan that touches a track but finds the same file sends nothing. A changed file goes to its old name; Minstrel never deletes there. An object the bucket refuses (a locked overwrite) is kept with its answer and goes again once the file changes or the settings are saved; three refusals in a row stop the night's music, since that is a key that may not write under music/. A folder copy is written under a .part name and renamed, keeping the file's time. The panel shows how much of the library is in, the refused files, and each run's music count and note. The verdict still turns on the database file. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,281 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
const (
|
||||
// The music's folder inside the destination, beside database/.
|
||||
musicDir = "music"
|
||||
// MusicBudget is how long one run may spend sending music. A library
|
||||
// bigger than a night's worth carries on from where it stopped on the
|
||||
// next run, so the first copy of a large library spans several nights.
|
||||
MusicBudget = 6 * time.Hour
|
||||
musicBatch = 200
|
||||
// refusedInARow stops a pass: one refusal is that file, several in a row
|
||||
// are the key or the bucket refusing the music as a whole.
|
||||
refusedInARow = 3
|
||||
)
|
||||
|
||||
// MusicProgress is how much of the library the backup holds as it is now.
|
||||
type MusicProgress struct {
|
||||
TotalFiles int64 `json:"total_files"`
|
||||
TotalBytes int64 `json:"total_bytes"`
|
||||
SentFiles int64 `json:"sent_files"`
|
||||
SentBytes int64 `json:"sent_bytes"`
|
||||
RefusedFiles int64 `json:"refused_files"`
|
||||
Refused []RefusedFile `json:"refused"`
|
||||
}
|
||||
|
||||
// RefusedFile is a library file the destination would not take, with its
|
||||
// answer. It is tried again once the file changes.
|
||||
type RefusedFile struct {
|
||||
Path string `json:"path"`
|
||||
Error string `json:"error"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// musicSent is what one run's music pass did.
|
||||
type musicSent struct {
|
||||
files int
|
||||
bytes int64
|
||||
note string
|
||||
}
|
||||
|
||||
// commonRoot is the deepest folder holding every library root, so the music's
|
||||
// paths in the backup are the same whichever root a file is under.
|
||||
func commonRoot(roots []string) string {
|
||||
if len(roots) == 0 {
|
||||
return ""
|
||||
}
|
||||
common := strings.Split(filepath.Clean(roots[0]), string(filepath.Separator))
|
||||
for _, r := range roots[1:] {
|
||||
parts := strings.Split(filepath.Clean(r), string(filepath.Separator))
|
||||
n := 0
|
||||
for n < len(common) && n < len(parts) && common[n] == parts[n] {
|
||||
n++
|
||||
}
|
||||
common = common[:n]
|
||||
}
|
||||
root := strings.Join(common, string(filepath.Separator))
|
||||
if root == "" {
|
||||
return string(filepath.Separator)
|
||||
}
|
||||
return root
|
||||
}
|
||||
|
||||
// musicPath is file's path inside music/, relative to root, with forward
|
||||
// slashes. A file outside root keeps its whole path.
|
||||
func musicPath(root, file string) string {
|
||||
rel, err := filepath.Rel(root, file)
|
||||
if root == "" || err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
|
||||
rel = strings.TrimPrefix(filepath.Clean(file), string(filepath.Separator))
|
||||
}
|
||||
return musicDir + "/" + filepath.ToSlash(rel)
|
||||
}
|
||||
|
||||
func (s *Service) musicProgress(ctx context.Context) (*MusicProgress, error) {
|
||||
q := dbq.New(s.pool)
|
||||
p, err := q.BackupMusicProgress(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("backup: read music progress: %w", err)
|
||||
}
|
||||
refused, err := q.RefusedBackupMusic(ctx, recentRuns)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("backup: list refused music: %w", err)
|
||||
}
|
||||
out := &MusicProgress{
|
||||
TotalFiles: p.TotalFiles, TotalBytes: p.TotalBytes,
|
||||
SentFiles: p.SentFiles, SentBytes: p.SentBytes,
|
||||
RefusedFiles: p.RefusedFiles, Refused: make([]RefusedFile, 0, len(refused)),
|
||||
}
|
||||
for _, r := range refused {
|
||||
out.Refused = append(out.Refused, RefusedFile{Path: r.FilePath, Error: r.Error, At: r.CheckedAt.Time})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// sendMusic sends library files the destination doesn't hold as they are now,
|
||||
// one at a time in path order, until all are sent or ctx ends. Every file is
|
||||
// recorded as it lands, so a run cut short loses nothing. It never deletes
|
||||
// anything at the destination.
|
||||
func (s *Service) sendMusic(ctx context.Context, st dbq.BackupSetting) musicSent {
|
||||
var out musicSent
|
||||
q := dbq.New(s.pool)
|
||||
var roots []string
|
||||
if s.Roots != nil {
|
||||
roots = s.Roots()
|
||||
}
|
||||
root := commonRoot(roots)
|
||||
var bucket Bucket
|
||||
if st.Destination == DestinationS3 {
|
||||
b, err := bucketOf(st)
|
||||
if err != nil {
|
||||
out.note = describe(err)
|
||||
return out
|
||||
}
|
||||
bucket = b
|
||||
}
|
||||
|
||||
after := ""
|
||||
refused := 0
|
||||
for {
|
||||
rows, err := q.PendingBackupMusic(ctx, dbq.PendingBackupMusicParams{AfterPath: after, MaxRows: musicBatch})
|
||||
if err != nil {
|
||||
out.note = musicStopped(ctx, err)
|
||||
return out
|
||||
}
|
||||
if len(rows) == 0 {
|
||||
return out
|
||||
}
|
||||
for _, r := range rows {
|
||||
after = r.FilePath
|
||||
if err := ctx.Err(); err != nil {
|
||||
out.note = musicStopped(ctx, err)
|
||||
return out
|
||||
}
|
||||
info, err := os.Stat(r.FilePath)
|
||||
if err != nil || !info.Mode().IsRegular() {
|
||||
// Gone or unreadable: the library scan settles that, and the
|
||||
// file is looked at again on the next run.
|
||||
continue
|
||||
}
|
||||
mtime := info.ModTime().Truncate(time.Microsecond)
|
||||
key := musicPath(root, r.FilePath)
|
||||
location := key
|
||||
if st.Destination == DestinationFolder {
|
||||
location = filepath.Join(st.Folder, filepath.FromSlash(key))
|
||||
}
|
||||
rec := dbq.RecordBackupMusicParams{
|
||||
TrackID: r.ID, Size: info.Size(),
|
||||
Mtime: pgtype.Timestamptz{Time: mtime, Valid: true},
|
||||
Location: location,
|
||||
}
|
||||
if r.SentSize != nil && *r.SentSize == info.Size() && r.SentMtime.Valid && r.SentMtime.Time.Equal(mtime) &&
|
||||
(r.SentError == nil || *r.SentError == "") {
|
||||
// Touched by a scan but the same file: nothing to send.
|
||||
if err := q.RecordBackupMusic(ctx, rec); err != nil {
|
||||
out.note = musicStopped(ctx, err)
|
||||
return out
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
if st.Destination == DestinationFolder {
|
||||
err = copyFile(r.FilePath, location, mtime)
|
||||
} else {
|
||||
err = putFile(ctx, bucket, key, r.FilePath, int64(st.UploadLimitKib)*1024)
|
||||
}
|
||||
switch {
|
||||
case err == nil:
|
||||
out.files++
|
||||
out.bytes += info.Size()
|
||||
refused = 0
|
||||
case ObjectRefused(err):
|
||||
rec.Error = strings.TrimPrefix(err.Error(), "backup: ")
|
||||
if refused++; refused >= refusedInARow {
|
||||
out.note = fmt.Sprintf("Music stopped: the destination refused %d files in a row (%s). "+
|
||||
"Check the key may write under %s/; refused files go again once you save the settings.",
|
||||
refused, rec.Error, musicDir)
|
||||
}
|
||||
case ctx.Err() == nil && errors.Is(err, errSourceRead):
|
||||
// The library file couldn't be read through: looked at again next run.
|
||||
continue
|
||||
default:
|
||||
out.note = musicStopped(ctx, err)
|
||||
return out
|
||||
}
|
||||
if err := q.RecordBackupMusic(ctx, rec); err != nil {
|
||||
out.note = musicStopped(ctx, err)
|
||||
return out
|
||||
}
|
||||
if out.note != "" {
|
||||
return out
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// musicStopped is why a music pass ended early, as the panel shows it.
|
||||
func musicStopped(ctx context.Context, err error) string {
|
||||
switch {
|
||||
case errors.Is(ctx.Err(), context.DeadlineExceeded):
|
||||
return fmt.Sprintf("Music stopped for tonight after %d hours; it carries on with the next backup.", int(MusicBudget/time.Hour))
|
||||
case ctx.Err() != nil:
|
||||
return "Music stopped because the server stopped; it carries on with the next backup."
|
||||
}
|
||||
return "Music stopped: " + describe(err) + " It carries on with the next backup."
|
||||
}
|
||||
|
||||
// errSourceRead marks a library file that could not be read.
|
||||
var errSourceRead = errors.New("backup: the library file could not be read")
|
||||
|
||||
// putFile uploads file to key. It reads the file twice, once to hash it
|
||||
// (S3 signs over the body's hash) and once to send it through the limit.
|
||||
func putFile(ctx context.Context, b Bucket, key, file string, perSecond int64) error {
|
||||
f, err := os.Open(file)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %v", errSourceRead, err)
|
||||
}
|
||||
defer func() { _ = f.Close() }()
|
||||
h := sha256.New()
|
||||
size, err := io.Copy(h, f)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %v", errSourceRead, err)
|
||||
}
|
||||
if _, err := f.Seek(0, io.SeekStart); err != nil {
|
||||
return fmt.Errorf("%w: %v", errSourceRead, err)
|
||||
}
|
||||
return b.Put(ctx, key, Throttle(ctx, f, perSecond), size, hex.EncodeToString(h.Sum(nil)))
|
||||
}
|
||||
|
||||
// copyFile copies file to dst under a hidden .part name first, so the folder
|
||||
// never holds half a file under the file's own name. The copy keeps the
|
||||
// file's modification time. The upload limit is the bucket's: a folder copy
|
||||
// runs at the disk's pace.
|
||||
func copyFile(file, dst string, mtime time.Time) error {
|
||||
in, err := os.Open(file)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %v", errSourceRead, err)
|
||||
}
|
||||
defer func() { _ = in.Close() }()
|
||||
if err := os.MkdirAll(filepath.Dir(dst), 0o700); err != nil {
|
||||
return fmt.Errorf("backup: create %s: %w", filepath.Dir(dst), err)
|
||||
}
|
||||
part := filepath.Join(filepath.Dir(dst), "."+filepath.Base(dst)+".part")
|
||||
out, err := os.OpenFile(part, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
|
||||
if err != nil {
|
||||
return fmt.Errorf("backup: create %s: %w", part, err)
|
||||
}
|
||||
_, err = io.Copy(out, in)
|
||||
if err == nil {
|
||||
err = out.Sync()
|
||||
}
|
||||
if cerr := out.Close(); err == nil {
|
||||
err = cerr
|
||||
}
|
||||
if err == nil {
|
||||
err = os.Chtimes(part, mtime, mtime)
|
||||
}
|
||||
if err == nil {
|
||||
err = os.Rename(part, dst)
|
||||
}
|
||||
if err != nil {
|
||||
_ = os.Remove(part)
|
||||
return fmt.Errorf("backup: write %s: %w", dst, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
package backup_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/backup"
|
||||
)
|
||||
|
||||
// library writes music files under a new library folder and adds a track for
|
||||
// each. It returns the folder.
|
||||
func library(t *testing.T, pool *pgxpool.Pool, files map[string]string) string {
|
||||
t.Helper()
|
||||
root := t.TempDir()
|
||||
artist := mustID(t, pool, `INSERT INTO artists (name, sort_name) VALUES ('Artist', 'artist') RETURNING id::text`)
|
||||
album := mustID(t, pool, `INSERT INTO albums (title, sort_title, artist_id) VALUES ('Album', 'album', $1) RETURNING id::text`, artist)
|
||||
for rel, body := range files {
|
||||
path := filepath.Join(root, filepath.FromSlash(rel))
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(body), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mustExec(t, pool, `INSERT INTO tracks (title, album_id, artist_id, duration_ms, file_path, file_size, file_format)
|
||||
VALUES ($1, $2, $3, 1000, $4, $5, 'flac')`, rel, album, artist, path, len(body))
|
||||
}
|
||||
return root
|
||||
}
|
||||
|
||||
func runOnce(t *testing.T, svc *backup.Service) backup.Run {
|
||||
t.Helper()
|
||||
if err := svc.Start(context.Background(), backup.TriggerManual); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitIdle(t, svc)
|
||||
st := status(t, svc)
|
||||
if len(st.Runs) == 0 || st.Runs[0].Status != "succeeded" {
|
||||
t.Fatalf("the run didn't succeed: %+v", st.Runs)
|
||||
}
|
||||
return st.Runs[0]
|
||||
}
|
||||
|
||||
func TestTheMusicGoesToTheFolderOnceAndAgainOnlyWhenItChanges_Integration(t *testing.T) {
|
||||
svc, pool := newService(t)
|
||||
ctx := context.Background()
|
||||
lib := library(t, pool, map[string]string{"A/one.flac": "one", "B/two.flac": "second"})
|
||||
svc.Roots = func() []string { return []string{lib} }
|
||||
dest := t.TempDir()
|
||||
if err := svc.Save(ctx, backup.Update{Enabled: true, Destination: backup.DestinationFolder, Folder: dest, RunHour: 3, IncludeMusic: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
first := runOnce(t, svc)
|
||||
if first.MusicFiles != 2 || first.MusicBytes != int64(len("one")+len("second")) || first.MusicNote != "" {
|
||||
t.Errorf("the first run sent %d files, %d bytes, note %q", first.MusicFiles, first.MusicBytes, first.MusicNote)
|
||||
}
|
||||
got, err := os.ReadFile(filepath.Join(dest, "music", "B", "two.flac"))
|
||||
if err != nil || string(got) != "second" {
|
||||
t.Errorf("music/B/two.flac = %q, %v", got, err)
|
||||
}
|
||||
src, _ := os.Stat(filepath.Join(lib, "A", "one.flac"))
|
||||
dst, err := os.Stat(filepath.Join(dest, "music", "A", "one.flac"))
|
||||
// The database keeps microseconds, so the copy's time is the file's to the µs.
|
||||
if err != nil || !dst.ModTime().Equal(src.ModTime().Truncate(time.Microsecond)) {
|
||||
t.Errorf("the copy's time = %v, want the file's %v (%v)", dst, src.ModTime(), err)
|
||||
}
|
||||
if p := status(t, svc).Music; p == nil || p.TotalFiles != 2 || p.SentFiles != 2 || p.RefusedFiles != 0 {
|
||||
t.Errorf("progress = %+v, want 2 of 2", p)
|
||||
}
|
||||
|
||||
if again := runOnce(t, svc); again.MusicFiles != 0 {
|
||||
t.Errorf("a second run sent %d files, want none", again.MusicFiles)
|
||||
}
|
||||
|
||||
// A scan that touched the track but found the same file sends nothing.
|
||||
mustExec(t, pool, `UPDATE tracks SET updated_at = now()`)
|
||||
if touched := runOnce(t, svc); touched.MusicFiles != 0 {
|
||||
t.Errorf("after a scan touched the tracks the run sent %d files, want none", touched.MusicFiles)
|
||||
}
|
||||
if p := status(t, svc).Music; p == nil || p.SentFiles != 2 {
|
||||
t.Errorf("progress after the scan = %+v, want 2 sent", p)
|
||||
}
|
||||
|
||||
// A changed file goes again, to the same place.
|
||||
path := filepath.Join(lib, "A", "one.flac")
|
||||
if err := os.WriteFile(path, []byte("one, retagged"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
later := time.Now().Add(time.Minute)
|
||||
if err := os.Chtimes(path, later, later); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mustExec(t, pool, `UPDATE tracks SET file_size = $1 WHERE file_path = $2`, len("one, retagged"), path)
|
||||
if changed := runOnce(t, svc); changed.MusicFiles != 1 {
|
||||
t.Errorf("after one file changed the run sent %d files, want 1", changed.MusicFiles)
|
||||
}
|
||||
if got, _ := os.ReadFile(filepath.Join(dest, "music", "A", "one.flac")); string(got) != "one, retagged" {
|
||||
t.Errorf("music/A/one.flac = %q, want the changed file", got)
|
||||
}
|
||||
if entries, _ := os.ReadDir(filepath.Join(dest, "music", "A")); len(entries) != 1 {
|
||||
t.Errorf("music/A holds %d entries, want just the file", len(entries))
|
||||
}
|
||||
}
|
||||
|
||||
// lockedBucket is a fakeBucket that refuses to overwrite an object, like a
|
||||
// bucket with object lock. With all set it refuses every music object, like
|
||||
// a key that may write only under database/.
|
||||
type lockedBucket struct {
|
||||
*fakeBucket
|
||||
all *atomic.Bool
|
||||
}
|
||||
|
||||
func (b lockedBucket) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method == http.MethodPut && strings.HasPrefix(r.URL.Path, "/minstrel/music/") {
|
||||
b.mu.Lock()
|
||||
_, held := b.puts[r.URL.Path]
|
||||
b.mu.Unlock()
|
||||
if held || (b.all != nil && b.all.Load()) {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
_, _ = w.Write([]byte(`<Error><Code>AccessDenied</Code><Message>object is locked</Message></Error>`))
|
||||
return
|
||||
}
|
||||
}
|
||||
b.fakeBucket.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
func TestTheMusicGoesToTheBucketAndARefusedOverwriteIsKept_Integration(t *testing.T) {
|
||||
svc, pool := newService(t)
|
||||
ctx := context.Background()
|
||||
fake := &fakeBucket{puts: map[string][]byte{}}
|
||||
srv := httptest.NewServer(lockedBucket{fakeBucket: fake})
|
||||
defer srv.Close()
|
||||
lib := library(t, pool, map[string]string{"A/one.flac": "one", "B/two.flac": "second"})
|
||||
svc.Roots = func() []string { return []string{lib} }
|
||||
u := bucketUpdate(srv.URL)
|
||||
u.IncludeMusic = true
|
||||
if err := svc.Save(ctx, u); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if first := runOnce(t, svc); first.MusicFiles != 2 {
|
||||
t.Fatalf("the first run sent %d files, want 2", first.MusicFiles)
|
||||
}
|
||||
fake.mu.Lock()
|
||||
body := string(fake.puts["/minstrel/music/A/one.flac"])
|
||||
fake.mu.Unlock()
|
||||
if body != "one" {
|
||||
t.Errorf("music/A/one.flac in the bucket = %q", body)
|
||||
}
|
||||
|
||||
path := filepath.Join(lib, "A", "one.flac")
|
||||
if err := os.WriteFile(path, []byte("one, retagged"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mustExec(t, pool, `UPDATE tracks SET file_size = $1, updated_at = now() WHERE file_path = $2`, len("one, retagged"), path)
|
||||
run := runOnce(t, svc)
|
||||
if run.MusicFiles != 0 || run.MusicNote != "" {
|
||||
t.Errorf("a refused overwrite: sent %d, note %q; want nothing sent and no note", run.MusicFiles, run.MusicNote)
|
||||
}
|
||||
p := status(t, svc).Music
|
||||
if p == nil || p.SentFiles != 1 || p.RefusedFiles != 1 || len(p.Refused) != 1 ||
|
||||
p.Refused[0].Path != path || !strings.Contains(p.Refused[0].Error, "AccessDenied") {
|
||||
t.Errorf("progress = %+v, want one sent and the overwrite refused", p)
|
||||
}
|
||||
|
||||
// Refused is settled until the file changes again.
|
||||
if again := runOnce(t, svc); again.MusicFiles != 0 {
|
||||
t.Errorf("the run after a refusal sent %d files, want none", again.MusicFiles)
|
||||
}
|
||||
if p := status(t, svc).Music; p == nil || p.RefusedFiles != 1 {
|
||||
t.Errorf("progress after the next run = %+v, want the refusal kept", p)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAKeyThatMayNotWriteTheMusicStopsThePassAndASaveTriesAgain_Integration(t *testing.T) {
|
||||
svc, pool := newService(t)
|
||||
ctx := context.Background()
|
||||
fake := &fakeBucket{puts: map[string][]byte{}}
|
||||
all := &atomic.Bool{}
|
||||
all.Store(true)
|
||||
srv := httptest.NewServer(lockedBucket{fakeBucket: fake, all: all})
|
||||
defer srv.Close()
|
||||
lib := library(t, pool, map[string]string{"a.flac": "a", "b.flac": "b", "c.flac": "c", "d.flac": "d"})
|
||||
svc.Roots = func() []string { return []string{lib} }
|
||||
u := bucketUpdate(srv.URL)
|
||||
u.IncludeMusic = true
|
||||
if err := svc.Save(ctx, u); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
run := runOnce(t, svc)
|
||||
if run.MusicFiles != 0 || !strings.Contains(run.MusicNote, "refused 3 files in a row") {
|
||||
t.Errorf("sent %d, note %q; want the pass stopped after three refusals", run.MusicFiles, run.MusicNote)
|
||||
}
|
||||
if p := status(t, svc).Music; p == nil || p.RefusedFiles != 3 {
|
||||
t.Errorf("progress = %+v, want three refused and the fourth not tried", p)
|
||||
}
|
||||
|
||||
// The key is fixed and the settings saved: the refused files go again.
|
||||
all.Store(false)
|
||||
if err := svc.Save(ctx, u); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again := runOnce(t, svc); again.MusicFiles != 4 || again.MusicNote != "" {
|
||||
t.Errorf("after the save: sent %d, note %q; want all four", again.MusicFiles, again.MusicNote)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestCommonRootIsTheFolderHoldingEveryLibraryRoot(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
roots []string
|
||||
want string
|
||||
}{
|
||||
{nil, ""},
|
||||
{[]string{"/music"}, "/music"},
|
||||
{[]string{"/music/"}, "/music"},
|
||||
{[]string{"/data/music/a", "/data/music/b"}, "/data/music"},
|
||||
{[]string{"/data/music", "/data/musicals"}, "/data"},
|
||||
{[]string{"/music", "/audiobooks"}, "/"},
|
||||
} {
|
||||
if got := commonRoot(c.roots); got != c.want {
|
||||
t.Errorf("commonRoot(%q) = %q, want %q", c.roots, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMusicPathIsTheFileUnderTheCommonRoot(t *testing.T) {
|
||||
for _, c := range []struct{ root, file, want string }{
|
||||
{"/data/music", "/data/music/a/Artist/01 Song.flac", "music/a/Artist/01 Song.flac"},
|
||||
{"/", "/music/x.mp3", "music/music/x.mp3"},
|
||||
{"/data/music", "/elsewhere/x.mp3", "music/elsewhere/x.mp3"},
|
||||
{"", "/music/x.mp3", "music/music/x.mp3"},
|
||||
} {
|
||||
if got := musicPath(c.root, c.file); got != c.want {
|
||||
t.Errorf("musicPath(%q, %q) = %q, want %q", c.root, c.file, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestObjectRefusedIsOneObjectNotTheWholeBucket(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{&Refusal{Status: http.StatusForbidden, Code: "AccessDenied"}, true},
|
||||
{&Refusal{Status: http.StatusConflict, Code: "ObjectLocked"}, true},
|
||||
{fmt.Errorf("put: %w", &Refusal{Status: http.StatusBadRequest, Code: "InvalidObjectName"}), true},
|
||||
{&Refusal{Status: http.StatusForbidden, Code: "SignatureDoesNotMatch"}, false},
|
||||
{&Refusal{Status: http.StatusNotFound, Code: "NoSuchBucket"}, false},
|
||||
{&Refusal{Status: http.StatusTooManyRequests}, false},
|
||||
{&Refusal{Status: http.StatusServiceUnavailable, Code: "SlowDown"}, false},
|
||||
{&Refusal{Status: http.StatusInternalServerError}, false},
|
||||
{errors.New("connection refused"), false},
|
||||
} {
|
||||
if got := ObjectRefused(c.err); got != c.want {
|
||||
t.Errorf("ObjectRefused(%v %+v) = %v, want %v", c.err, c.err, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
+33
-3
@@ -171,6 +171,17 @@ func (b Bucket) List(ctx context.Context, prefix string, limit int) ([]Object, e
|
||||
}
|
||||
}
|
||||
|
||||
// Refusal is an S3 error response: its HTTP status, S3's code, and the
|
||||
// reason in words. It matches ErrS3.
|
||||
type Refusal struct {
|
||||
Status int
|
||||
Code string
|
||||
text string
|
||||
}
|
||||
|
||||
func (r *Refusal) Error() string { return r.text }
|
||||
func (r *Refusal) Unwrap() error { return ErrS3 }
|
||||
|
||||
// refusal turns an S3 error response into words an admin can act on.
|
||||
func refusal(resp *http.Response) error {
|
||||
var e struct {
|
||||
@@ -185,14 +196,33 @@ func refusal(resp *http.Response) error {
|
||||
"NoSuchBucket": "there is no bucket by that name",
|
||||
"SlowDown": "the server asks to slow down",
|
||||
}[e.Code]
|
||||
r := &Refusal{Status: resp.StatusCode, Code: e.Code}
|
||||
switch {
|
||||
case hint != "":
|
||||
return fmt.Errorf("%w: HTTP %d %s: %s", ErrS3, resp.StatusCode, e.Code, hint)
|
||||
r.text = fmt.Sprintf("%v: HTTP %d %s: %s", ErrS3, resp.StatusCode, e.Code, hint)
|
||||
case e.Code != "":
|
||||
return fmt.Errorf("%w: HTTP %d %s: %s", ErrS3, resp.StatusCode, e.Code, e.Message)
|
||||
r.text = fmt.Sprintf("%v: HTTP %d %s: %s", ErrS3, resp.StatusCode, e.Code, e.Message)
|
||||
default:
|
||||
return fmt.Errorf("%w: HTTP %d", ErrS3, resp.StatusCode)
|
||||
r.text = fmt.Sprintf("%v: HTTP %d", ErrS3, resp.StatusCode)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// ObjectRefused reports whether err is the bucket refusing one object (an
|
||||
// overwrite a write-once target won't take, a name it won't store) rather
|
||||
// than refusing everything: bad keys, a missing bucket, a request to slow
|
||||
// down, or a server error all stop a run instead.
|
||||
func ObjectRefused(err error) bool {
|
||||
var r *Refusal
|
||||
if !errors.As(err, &r) || r.Status/100 != 4 || r.Status == http.StatusTooManyRequests {
|
||||
return false
|
||||
}
|
||||
switch r.Code {
|
||||
case "SignatureDoesNotMatch", "InvalidAccessKeyId", "NoSuchBucket", "SlowDown",
|
||||
"RequestTimeTooSkewed", "ExpiredToken", "InvalidToken":
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func drain(resp *http.Response) {
|
||||
|
||||
@@ -94,6 +94,7 @@ type Config struct {
|
||||
HasSecretKey bool `json:"has_secret_key"`
|
||||
RunHour int `json:"run_hour"`
|
||||
UploadLimitKiB int `json:"upload_limit_kib"`
|
||||
IncludeMusic bool `json:"include_music"`
|
||||
}
|
||||
|
||||
// Update is a save from the admin card. An empty SecretKey keeps the stored
|
||||
@@ -108,6 +109,7 @@ type Update struct {
|
||||
SecretKey string `json:"secret_key"`
|
||||
RunHour int `json:"run_hour"`
|
||||
UploadLimitKiB int `json:"upload_limit_kib"`
|
||||
IncludeMusic bool `json:"include_music"`
|
||||
}
|
||||
|
||||
// Run is one attempt at a backup.
|
||||
@@ -120,6 +122,10 @@ type Run struct {
|
||||
Location string `json:"location"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
Error string `json:"error"`
|
||||
// What the run sent of the music, and why that stopped early, if it did.
|
||||
MusicFiles int `json:"music_files"`
|
||||
MusicBytes int64 `json:"music_bytes"`
|
||||
MusicNote string `json:"music_note"`
|
||||
}
|
||||
|
||||
// Status is everything the admin panel shows: a verdict, what stands in its
|
||||
@@ -136,6 +142,9 @@ type Status struct {
|
||||
NextRun *time.Time `json:"next_run"`
|
||||
RunLimitSeconds int `json:"run_limit_seconds"`
|
||||
Runs []Run `json:"runs"`
|
||||
// Music is how much of the library the backup holds, when it includes
|
||||
// the music.
|
||||
Music *MusicProgress `json:"music"`
|
||||
}
|
||||
|
||||
// TestResult is what a destination test found: how many backups are there
|
||||
@@ -152,6 +161,9 @@ type Service struct {
|
||||
logger *slog.Logger
|
||||
build string
|
||||
now func() time.Time
|
||||
// Roots lists the library's folders. The music's paths in the backup are
|
||||
// relative to the folder holding them all.
|
||||
Roots func() []string
|
||||
|
||||
base context.Context
|
||||
cancel context.CancelFunc
|
||||
@@ -188,6 +200,7 @@ func view(st dbq.BackupSetting) Config {
|
||||
HasSecretKey: st.SecretKey != "",
|
||||
RunHour: int(st.RunHour),
|
||||
UploadLimitKiB: int(st.UploadLimitKib),
|
||||
IncludeMusic: st.IncludeMusic,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -231,6 +244,7 @@ func merge(st dbq.BackupSetting, u Update) (dbq.BackupSetting, error) {
|
||||
st.AccessKey = strings.TrimSpace(u.AccessKey)
|
||||
st.RunHour = int16(u.RunHour)
|
||||
st.UploadLimitKib = int32(u.UploadLimitKiB)
|
||||
st.IncludeMusic = u.IncludeMusic
|
||||
if u.SecretKey != "" {
|
||||
sealed, err := secret.Seal(secret.BackupSecretKey.Purpose(), "", u.SecretKey)
|
||||
if err != nil {
|
||||
@@ -262,6 +276,7 @@ func (s *Service) Save(ctx context.Context, u Update) error {
|
||||
SecretKey: st.SecretKey,
|
||||
RunHour: st.RunHour,
|
||||
UploadLimitKib: st.UploadLimitKib,
|
||||
IncludeMusic: st.IncludeMusic,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -394,12 +409,17 @@ func (s *Service) Status(ctx context.Context) (Status, error) {
|
||||
Problems: probs,
|
||||
LastSuccess: ptr(t.lastSuccess),
|
||||
FailingSince: ptr(t.failingSince),
|
||||
RunLimitSeconds: int(RunLimit / time.Second),
|
||||
RunLimitSeconds: int(runLimit(st) / time.Second),
|
||||
Runs: make([]Run, 0, len(rows)),
|
||||
}
|
||||
if out.Problems == nil {
|
||||
out.Problems = []string{}
|
||||
}
|
||||
if st.IncludeMusic {
|
||||
if out.Music, err = s.musicProgress(ctx); err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
}
|
||||
s.mu.Lock()
|
||||
out.RunningSince = ptr(s.runningSince)
|
||||
s.mu.Unlock()
|
||||
@@ -421,6 +441,9 @@ func (s *Service) Status(ctx context.Context) (Status, error) {
|
||||
Location: r.Location,
|
||||
Bytes: r.Bytes,
|
||||
Error: r.Error,
|
||||
MusicFiles: int(r.MusicFiles),
|
||||
MusicBytes: r.MusicBytes,
|
||||
MusicNote: r.MusicNote,
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
@@ -545,6 +568,15 @@ func (s *Service) Running() time.Time {
|
||||
return s.runningSince
|
||||
}
|
||||
|
||||
// runLimit is the longest a run may take: the database file, and the
|
||||
// music's nightly budget when the backup includes the music.
|
||||
func runLimit(st dbq.BackupSetting) time.Duration {
|
||||
if st.IncludeMusic {
|
||||
return RunLimit + MusicBudget
|
||||
}
|
||||
return RunLimit
|
||||
}
|
||||
|
||||
func (s *Service) run(trigger string, st dbq.BackupSetting) {
|
||||
ctx, cancel := context.WithTimeout(s.base, RunLimit)
|
||||
defer cancel()
|
||||
@@ -555,17 +587,29 @@ func (s *Service) run(trigger string, st dbq.BackupSetting) {
|
||||
return
|
||||
}
|
||||
location, size, err := s.write(ctx, st, started.StartedAt.Time)
|
||||
// The music goes only after the database file has landed, which also
|
||||
// proves the destination takes writes.
|
||||
var music musicSent
|
||||
if err == nil && st.IncludeMusic {
|
||||
mctx, mcancel := context.WithTimeout(s.base, MusicBudget)
|
||||
music = s.sendMusic(mctx, st)
|
||||
mcancel()
|
||||
}
|
||||
|
||||
// The finish is recorded on its own context, so a run stopped by a
|
||||
// shutdown or its time limit still says how it ended.
|
||||
fctx, fcancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer fcancel()
|
||||
fin := dbq.FinishBackupRunParams{ID: started.ID, Status: "succeeded", Location: location, Bytes: size}
|
||||
fin := dbq.FinishBackupRunParams{
|
||||
ID: started.ID, Status: "succeeded", Location: location, Bytes: size,
|
||||
MusicFiles: int32(music.files), MusicBytes: music.bytes, MusicNote: music.note,
|
||||
}
|
||||
if err != nil {
|
||||
fin.Status, fin.Error = "failed", describe(err)
|
||||
s.logger.Warn("backup failed", "trigger", trigger, "err", err)
|
||||
} else {
|
||||
s.logger.Info("backup written", "trigger", trigger, "location", location, "bytes", size)
|
||||
s.logger.Info("backup written", "trigger", trigger, "location", location, "bytes", size,
|
||||
"music_files", music.files, "music_bytes", music.bytes, "music_note", music.note)
|
||||
}
|
||||
if err := q.FinishBackupRun(fctx, fin); err != nil {
|
||||
s.logger.Error("backup: record the run's end", "err", err)
|
||||
|
||||
@@ -147,8 +147,9 @@ var Excluded = map[string]string{
|
||||
"scan_runs": "a log of past library scans",
|
||||
"diagnostic_events": "debug-mode diagnostics",
|
||||
|
||||
"backup_settings": "holds the bucket's secret key; point the restored server at its destination again",
|
||||
"backup_runs": "this server's own backup history",
|
||||
"backup_settings": "holds the bucket's secret key; point the restored server at its destination again",
|
||||
"backup_runs": "this server's own backup history",
|
||||
"backup_music_files": "which music files this server has sent; the files themselves are beside the database in the destination",
|
||||
}
|
||||
|
||||
// carried reports whether a table is in Tables.
|
||||
|
||||
@@ -11,6 +11,39 @@ import (
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
const backupMusicProgress = `-- name: BackupMusicProgress :one
|
||||
SELECT count(*)::bigint AS total_files,
|
||||
COALESCE(sum(t.file_size), 0)::bigint AS total_bytes,
|
||||
count(*) FILTER (WHERE m.error = '' AND m.size = t.file_size AND t.updated_at <= m.checked_at)::bigint AS sent_files,
|
||||
COALESCE(sum(t.file_size) FILTER (WHERE m.error = '' AND m.size = t.file_size AND t.updated_at <= m.checked_at), 0)::bigint AS sent_bytes,
|
||||
count(*) FILTER (WHERE m.error <> '' AND m.size = t.file_size AND t.updated_at <= m.checked_at)::bigint AS refused_files
|
||||
FROM tracks t
|
||||
LEFT JOIN backup_music_files m ON m.track_id = t.id
|
||||
WHERE t.missing_since IS NULL
|
||||
`
|
||||
|
||||
type BackupMusicProgressRow struct {
|
||||
TotalFiles int64
|
||||
TotalBytes int64
|
||||
SentFiles int64
|
||||
SentBytes int64
|
||||
RefusedFiles int64
|
||||
}
|
||||
|
||||
// How much of the library the backup holds as it is now.
|
||||
func (q *Queries) BackupMusicProgress(ctx context.Context) (BackupMusicProgressRow, error) {
|
||||
row := q.db.QueryRow(ctx, backupMusicProgress)
|
||||
var i BackupMusicProgressRow
|
||||
err := row.Scan(
|
||||
&i.TotalFiles,
|
||||
&i.TotalBytes,
|
||||
&i.SentFiles,
|
||||
&i.SentBytes,
|
||||
&i.RefusedFiles,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const backupRunTimes = `-- name: BackupRunTimes :one
|
||||
SELECT
|
||||
COALESCE((SELECT max(started_at) FROM backup_runs WHERE status = 'succeeded'),
|
||||
@@ -61,16 +94,22 @@ UPDATE backup_runs
|
||||
finished_at = now(),
|
||||
location = $3,
|
||||
bytes = $4,
|
||||
error = $5
|
||||
error = $5,
|
||||
music_files = $6,
|
||||
music_bytes = $7,
|
||||
music_note = $8
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type FinishBackupRunParams struct {
|
||||
ID pgtype.UUID
|
||||
Status string
|
||||
Location string
|
||||
Bytes int64
|
||||
Error string
|
||||
ID pgtype.UUID
|
||||
Status string
|
||||
Location string
|
||||
Bytes int64
|
||||
Error string
|
||||
MusicFiles int32
|
||||
MusicBytes int64
|
||||
MusicNote string
|
||||
}
|
||||
|
||||
func (q *Queries) FinishBackupRun(ctx context.Context, arg FinishBackupRunParams) error {
|
||||
@@ -80,13 +119,16 @@ func (q *Queries) FinishBackupRun(ctx context.Context, arg FinishBackupRunParams
|
||||
arg.Location,
|
||||
arg.Bytes,
|
||||
arg.Error,
|
||||
arg.MusicFiles,
|
||||
arg.MusicBytes,
|
||||
arg.MusicNote,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const getBackupSettings = `-- name: GetBackupSettings :one
|
||||
|
||||
SELECT id, enabled, destination, folder, endpoint, bucket, access_key, secret_key, run_hour, upload_limit_kib, updated_at FROM backup_settings WHERE id = true
|
||||
SELECT id, enabled, destination, folder, endpoint, bucket, access_key, secret_key, run_hour, upload_limit_kib, updated_at, include_music FROM backup_settings WHERE id = true
|
||||
`
|
||||
|
||||
// Minstrel's own backup: its settings and its runs.
|
||||
@@ -105,12 +147,14 @@ func (q *Queries) GetBackupSettings(ctx context.Context) (BackupSetting, error)
|
||||
&i.RunHour,
|
||||
&i.UploadLimitKib,
|
||||
&i.UpdatedAt,
|
||||
&i.IncludeMusic,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const listBackupRuns = `-- name: ListBackupRuns :many
|
||||
SELECT id, trigger, status, started_at, finished_at, location, bytes, error
|
||||
SELECT id, trigger, status, started_at, finished_at, location, bytes, error,
|
||||
music_files, music_bytes, music_note
|
||||
FROM backup_runs
|
||||
ORDER BY started_at DESC
|
||||
LIMIT $1
|
||||
@@ -134,6 +178,69 @@ func (q *Queries) ListBackupRuns(ctx context.Context, limit int32) ([]BackupRun,
|
||||
&i.Location,
|
||||
&i.Bytes,
|
||||
&i.Error,
|
||||
&i.MusicFiles,
|
||||
&i.MusicBytes,
|
||||
&i.MusicNote,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const pendingBackupMusic = `-- name: PendingBackupMusic :many
|
||||
SELECT t.id, t.file_path, t.file_size,
|
||||
m.size AS sent_size, m.mtime AS sent_mtime, m.error AS sent_error
|
||||
FROM tracks t
|
||||
LEFT JOIN backup_music_files m ON m.track_id = t.id
|
||||
WHERE t.missing_since IS NULL
|
||||
AND t.file_path > $1::text
|
||||
AND (m.track_id IS NULL OR m.size <> t.file_size OR t.updated_at > m.checked_at
|
||||
OR (m.error <> '' AND m.checked_at < (SELECT updated_at FROM backup_settings)))
|
||||
ORDER BY t.file_path
|
||||
LIMIT $2
|
||||
`
|
||||
|
||||
type PendingBackupMusicParams struct {
|
||||
AfterPath string
|
||||
MaxRows int32
|
||||
}
|
||||
|
||||
type PendingBackupMusicRow struct {
|
||||
ID pgtype.UUID
|
||||
FilePath string
|
||||
FileSize int64
|
||||
SentSize *int64
|
||||
SentMtime pgtype.Timestamptz
|
||||
SentError *string
|
||||
}
|
||||
|
||||
// Library files the backup may not hold as they are now, in path order from
|
||||
// after_path, so a night that runs out of time carries on where it stopped:
|
||||
// never sent, a different size, touched by a scan since they were last
|
||||
// checked, or refused before the settings were last saved (a fixed key gets
|
||||
// another go). The caller compares the file's own size and mtime before
|
||||
// sending.
|
||||
func (q *Queries) PendingBackupMusic(ctx context.Context, arg PendingBackupMusicParams) ([]PendingBackupMusicRow, error) {
|
||||
rows, err := q.db.Query(ctx, pendingBackupMusic, arg.AfterPath, arg.MaxRows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []PendingBackupMusicRow
|
||||
for rows.Next() {
|
||||
var i PendingBackupMusicRow
|
||||
if err := rows.Scan(
|
||||
&i.ID,
|
||||
&i.FilePath,
|
||||
&i.FileSize,
|
||||
&i.SentSize,
|
||||
&i.SentMtime,
|
||||
&i.SentError,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -157,6 +264,68 @@ func (q *Queries) PruneBackupRuns(ctx context.Context, startedAt pgtype.Timestam
|
||||
return result.RowsAffected(), nil
|
||||
}
|
||||
|
||||
const recordBackupMusic = `-- name: RecordBackupMusic :exec
|
||||
INSERT INTO backup_music_files (track_id, size, mtime, location, error)
|
||||
VALUES ($1, $2, $3, $4, $5)
|
||||
ON CONFLICT (track_id) DO UPDATE
|
||||
SET size = EXCLUDED.size, mtime = EXCLUDED.mtime, location = EXCLUDED.location,
|
||||
error = EXCLUDED.error, checked_at = now()
|
||||
`
|
||||
|
||||
type RecordBackupMusicParams struct {
|
||||
TrackID pgtype.UUID
|
||||
Size int64
|
||||
Mtime pgtype.Timestamptz
|
||||
Location string
|
||||
Error string
|
||||
}
|
||||
|
||||
func (q *Queries) RecordBackupMusic(ctx context.Context, arg RecordBackupMusicParams) error {
|
||||
_, err := q.db.Exec(ctx, recordBackupMusic,
|
||||
arg.TrackID,
|
||||
arg.Size,
|
||||
arg.Mtime,
|
||||
arg.Location,
|
||||
arg.Error,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const refusedBackupMusic = `-- name: RefusedBackupMusic :many
|
||||
SELECT t.file_path, m.error, m.checked_at
|
||||
FROM backup_music_files m
|
||||
JOIN tracks t ON t.id = m.track_id
|
||||
WHERE m.error <> '' AND m.size = t.file_size AND t.updated_at <= m.checked_at
|
||||
ORDER BY m.checked_at DESC
|
||||
LIMIT $1
|
||||
`
|
||||
|
||||
type RefusedBackupMusicRow struct {
|
||||
FilePath string
|
||||
Error string
|
||||
CheckedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
func (q *Queries) RefusedBackupMusic(ctx context.Context, limit int32) ([]RefusedBackupMusicRow, error) {
|
||||
rows, err := q.db.Query(ctx, refusedBackupMusic, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []RefusedBackupMusicRow
|
||||
for rows.Next() {
|
||||
var i RefusedBackupMusicRow
|
||||
if err := rows.Scan(&i.FilePath, &i.Error, &i.CheckedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const saveBackupSettings = `-- name: SaveBackupSettings :exec
|
||||
UPDATE backup_settings
|
||||
SET enabled = $1,
|
||||
@@ -168,6 +337,7 @@ UPDATE backup_settings
|
||||
secret_key = $7,
|
||||
run_hour = $8,
|
||||
upload_limit_kib = $9,
|
||||
include_music = $10,
|
||||
updated_at = now()
|
||||
WHERE id = true
|
||||
`
|
||||
@@ -182,6 +352,7 @@ type SaveBackupSettingsParams struct {
|
||||
SecretKey string
|
||||
RunHour int16
|
||||
UploadLimitKib int32
|
||||
IncludeMusic bool
|
||||
}
|
||||
|
||||
func (q *Queries) SaveBackupSettings(ctx context.Context, arg SaveBackupSettingsParams) error {
|
||||
@@ -195,6 +366,7 @@ func (q *Queries) SaveBackupSettings(ctx context.Context, arg SaveBackupSettings
|
||||
arg.SecretKey,
|
||||
arg.RunHour,
|
||||
arg.UploadLimitKib,
|
||||
arg.IncludeMusic,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -305,6 +305,15 @@ type AuditLog struct {
|
||||
CreatedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
type BackupMusicFile struct {
|
||||
TrackID pgtype.UUID
|
||||
Size int64
|
||||
Mtime pgtype.Timestamptz
|
||||
Location string
|
||||
CheckedAt pgtype.Timestamptz
|
||||
Error string
|
||||
}
|
||||
|
||||
type BackupRun struct {
|
||||
ID pgtype.UUID
|
||||
Trigger string
|
||||
@@ -314,6 +323,9 @@ type BackupRun struct {
|
||||
Location string
|
||||
Bytes int64
|
||||
Error string
|
||||
MusicFiles int32
|
||||
MusicBytes int64
|
||||
MusicNote string
|
||||
}
|
||||
|
||||
type BackupSetting struct {
|
||||
@@ -328,6 +340,7 @@ type BackupSetting struct {
|
||||
RunHour int16
|
||||
UploadLimitKib int32
|
||||
UpdatedAt pgtype.Timestamptz
|
||||
IncludeMusic bool
|
||||
}
|
||||
|
||||
type CandidateArtistTag struct {
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
DROP TABLE IF EXISTS backup_music_files;
|
||||
ALTER TABLE backup_runs
|
||||
DROP COLUMN IF EXISTS music_note,
|
||||
DROP COLUMN IF EXISTS music_bytes,
|
||||
DROP COLUMN IF EXISTS music_files;
|
||||
ALTER TABLE backup_settings DROP COLUMN IF EXISTS include_music;
|
||||
@@ -0,0 +1,36 @@
|
||||
-- 0100_backup_music.up.sql — the music files in Minstrel's own backup.
|
||||
--
|
||||
-- Until now a backup held the database only. With include_music on, each
|
||||
-- library file is also sent once to music/<path> in the same destination,
|
||||
-- and sent again only when it changes.
|
||||
--
|
||||
-- backup_settings.include_music off until an admin switches it on.
|
||||
-- backup_music_files one row per track sent (or refused): the
|
||||
-- file's size and mtime when it was sent, so
|
||||
-- a later night sends only what is new or
|
||||
-- changed. A track the scan has touched since
|
||||
-- checked_at is looked at again; it goes up
|
||||
-- again only if the file itself changed, so a
|
||||
-- full rescan doesn't resend the library. A
|
||||
-- refusal is kept with its reason and tried
|
||||
-- again once the file changes.
|
||||
-- backup_runs.music_* how much a run sent, and why it stopped.
|
||||
ALTER TABLE backup_settings ADD COLUMN include_music boolean NOT NULL DEFAULT false;
|
||||
|
||||
ALTER TABLE backup_runs
|
||||
ADD COLUMN music_files integer NOT NULL DEFAULT 0,
|
||||
ADD COLUMN music_bytes bigint NOT NULL DEFAULT 0,
|
||||
ADD COLUMN music_note text NOT NULL DEFAULT '';
|
||||
|
||||
CREATE TABLE backup_music_files (
|
||||
track_id uuid PRIMARY KEY REFERENCES tracks (id) ON DELETE CASCADE,
|
||||
-- The file as it was sent.
|
||||
size bigint NOT NULL,
|
||||
mtime timestamptz NOT NULL,
|
||||
-- Where it went: the object key, or the file's path in the folder.
|
||||
location text NOT NULL,
|
||||
-- When the file was last sent or found unchanged.
|
||||
checked_at timestamptz NOT NULL DEFAULT now(),
|
||||
-- Empty when it was sent; otherwise why the destination refused it.
|
||||
error text NOT NULL DEFAULT ''
|
||||
);
|
||||
@@ -14,6 +14,7 @@ UPDATE backup_settings
|
||||
secret_key = $7,
|
||||
run_hour = $8,
|
||||
upload_limit_kib = $9,
|
||||
include_music = $10,
|
||||
updated_at = now()
|
||||
WHERE id = true;
|
||||
|
||||
@@ -27,11 +28,15 @@ UPDATE backup_runs
|
||||
finished_at = now(),
|
||||
location = $3,
|
||||
bytes = $4,
|
||||
error = $5
|
||||
error = $5,
|
||||
music_files = $6,
|
||||
music_bytes = $7,
|
||||
music_note = $8
|
||||
WHERE id = $1;
|
||||
|
||||
-- name: ListBackupRuns :many
|
||||
SELECT id, trigger, status, started_at, finished_at, location, bytes, error
|
||||
SELECT id, trigger, status, started_at, finished_at, location, bytes, error,
|
||||
music_files, music_bytes, music_note
|
||||
FROM backup_runs
|
||||
ORDER BY started_at DESC
|
||||
LIMIT $1;
|
||||
@@ -59,3 +64,47 @@ UPDATE backup_runs
|
||||
|
||||
-- name: PruneBackupRuns :execrows
|
||||
DELETE FROM backup_runs WHERE status <> 'running' AND started_at < $1;
|
||||
|
||||
-- name: PendingBackupMusic :many
|
||||
-- Library files the backup may not hold as they are now, in path order from
|
||||
-- after_path, so a night that runs out of time carries on where it stopped:
|
||||
-- never sent, a different size, touched by a scan since they were last
|
||||
-- checked, or refused before the settings were last saved (a fixed key gets
|
||||
-- another go). The caller compares the file's own size and mtime before
|
||||
-- sending.
|
||||
SELECT t.id, t.file_path, t.file_size,
|
||||
m.size AS sent_size, m.mtime AS sent_mtime, m.error AS sent_error
|
||||
FROM tracks t
|
||||
LEFT JOIN backup_music_files m ON m.track_id = t.id
|
||||
WHERE t.missing_since IS NULL
|
||||
AND t.file_path > sqlc.arg(after_path)::text
|
||||
AND (m.track_id IS NULL OR m.size <> t.file_size OR t.updated_at > m.checked_at
|
||||
OR (m.error <> '' AND m.checked_at < (SELECT updated_at FROM backup_settings)))
|
||||
ORDER BY t.file_path
|
||||
LIMIT sqlc.arg(max_rows);
|
||||
|
||||
-- name: RecordBackupMusic :exec
|
||||
INSERT INTO backup_music_files (track_id, size, mtime, location, error)
|
||||
VALUES ($1, $2, $3, $4, $5)
|
||||
ON CONFLICT (track_id) DO UPDATE
|
||||
SET size = EXCLUDED.size, mtime = EXCLUDED.mtime, location = EXCLUDED.location,
|
||||
error = EXCLUDED.error, checked_at = now();
|
||||
|
||||
-- name: BackupMusicProgress :one
|
||||
-- How much of the library the backup holds as it is now.
|
||||
SELECT count(*)::bigint AS total_files,
|
||||
COALESCE(sum(t.file_size), 0)::bigint AS total_bytes,
|
||||
count(*) FILTER (WHERE m.error = '' AND m.size = t.file_size AND t.updated_at <= m.checked_at)::bigint AS sent_files,
|
||||
COALESCE(sum(t.file_size) FILTER (WHERE m.error = '' AND m.size = t.file_size AND t.updated_at <= m.checked_at), 0)::bigint AS sent_bytes,
|
||||
count(*) FILTER (WHERE m.error <> '' AND m.size = t.file_size AND t.updated_at <= m.checked_at)::bigint AS refused_files
|
||||
FROM tracks t
|
||||
LEFT JOIN backup_music_files m ON m.track_id = t.id
|
||||
WHERE t.missing_since IS NULL;
|
||||
|
||||
-- name: RefusedBackupMusic :many
|
||||
SELECT t.file_path, m.error, m.checked_at
|
||||
FROM backup_music_files m
|
||||
JOIN tracks t ON t.id = m.track_id
|
||||
WHERE m.error <> '' AND m.size = t.file_size AND t.updated_at <= m.checked_at
|
||||
ORDER BY m.checked_at DESC
|
||||
LIMIT $1;
|
||||
|
||||
@@ -187,10 +187,11 @@ func ResetDB(t *testing.T, pool *pgxpool.Pool) {
|
||||
// The server's own backup: no runs, and settings back to off.
|
||||
if _, err := pool.Exec(ctx, `
|
||||
DELETE FROM backup_runs;
|
||||
DELETE FROM backup_music_files;
|
||||
UPDATE backup_settings
|
||||
SET enabled = DEFAULT, destination = DEFAULT, folder = DEFAULT, endpoint = DEFAULT,
|
||||
bucket = DEFAULT, access_key = DEFAULT, secret_key = DEFAULT, run_hour = DEFAULT,
|
||||
upload_limit_kib = DEFAULT, updated_at = DEFAULT`,
|
||||
upload_limit_kib = DEFAULT, include_music = DEFAULT, updated_at = DEFAULT`,
|
||||
); err != nil {
|
||||
t.Fatalf("dbtest.ResetDB reset backup settings: %v", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user