feat: Minstrel backs itself up nightly to a folder or an S3 bucket, off until an admin switches it on (#6773)
release / web (push) Failing after 15s
release / govulncheck (push) Successful in 16s
release / go (push) Failing after 41s
release / integration (push) Failing after 3m29s
release / android (push) Successful in 4m7s
release / Attach APK to the Release (tag releases only) (push) Canceled after 0s
release / Build + push container image (push) Canceled after 0s
release / Verify release artifacts (tag releases only) (push) Canceled after 0s
release / Build signed APK (push) Canceled after 5m34s
release / web (push) Failing after 15s
release / govulncheck (push) Successful in 16s
release / go (push) Failing after 41s
release / integration (push) Failing after 3m29s
release / android (push) Successful in 4m7s
release / Attach APK to the Release (tag releases only) (push) Canceled after 0s
release / Build + push container image (push) Canceled after 0s
release / Verify release artifacts (tag releases only) (push) Canceled after 0s
release / Build signed APK (push) Canceled after 5m34s
Admin › Backups shows a verdict first (off, not set up, waiting, backed up, overdue, failing since when), Back up now, Test destination and the latest runs, with the settings beside them. - backup.Service: one run at a time, bounded at 30 minutes; the schedule ticks every 5 minutes, catches up a slot missed while the server was down, and retries a failure after an hour and then after as long as it has been failing. Runs are recorded in backup_runs, pruned after 90 days; a run cut off by a shutdown is marked failed at the next start. - A folder destination writes database/minstrel-<UTC stamp>.tar and refuses a folder that isn't there. An S3 destination stages the file, hashes it, and uploads it under database/ through a throttled reader (4 MiB/s by default, 0 for no limit). - Test destination checks the form as typed and only lists. - The bucket secret is sealed like every stored credential; backup_settings and backup_runs are left out of the backup itself. - The panel polls only while a run is going, capped at the run limit plus five minutes. The music switch named in the task is left to #6774, so no control ships that does nothing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,94 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/apierror"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/backup"
|
||||
)
|
||||
|
||||
// GET /api/admin/backups — the backup's verdict, settings and recent runs.
|
||||
func (h *handlers) handleGetBackups(w http.ResponseWriter, r *http.Request) {
|
||||
st, err := h.backups.Status(r.Context())
|
||||
if err != nil {
|
||||
writeErrWithLog(w, h.logger, "admin backups: status", apierror.Internal(err))
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, st)
|
||||
}
|
||||
|
||||
func decodeBackupUpdate(w http.ResponseWriter, r *http.Request) (backup.Update, bool) {
|
||||
var u backup.Update
|
||||
if err := json.NewDecoder(r.Body).Decode(&u); err != nil {
|
||||
writeErr(w, apierror.BadRequest("invalid_body", ""))
|
||||
return u, false
|
||||
}
|
||||
return u, true
|
||||
}
|
||||
|
||||
// fieldErr writes a setting that can't be kept as a 400 whose message says
|
||||
// which setting and why, and reports whether err was one.
|
||||
func fieldErr(w http.ResponseWriter, err error) bool {
|
||||
var fe *backup.FieldError
|
||||
if !errors.As(err, &fe) {
|
||||
return false
|
||||
}
|
||||
writeErr(w, apierror.BadRequest("invalid_backup_setting", fe.Message))
|
||||
return true
|
||||
}
|
||||
|
||||
// PUT /api/admin/backups — save the settings. An empty secret_key keeps the
|
||||
// stored one. Returns the new status.
|
||||
func (h *handlers) handleUpdateBackups(w http.ResponseWriter, r *http.Request) {
|
||||
u, ok := decodeBackupUpdate(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if err := h.backups.Save(r.Context(), u); err != nil {
|
||||
if !fieldErr(w, err) {
|
||||
writeErrWithLog(w, h.logger, "admin backups: save", apierror.Internal(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
h.handleGetBackups(w, r)
|
||||
}
|
||||
|
||||
// POST /api/admin/backups/test — check the destination in the body without
|
||||
// saving it, by listing the backups already there. Nothing is written.
|
||||
func (h *handlers) handleTestBackups(w http.ResponseWriter, r *http.Request) {
|
||||
u, ok := decodeBackupUpdate(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
res, err := h.backups.Test(r.Context(), u)
|
||||
if err != nil {
|
||||
if fieldErr(w, err) {
|
||||
return
|
||||
}
|
||||
msg := strings.TrimPrefix(err.Error(), "backup: ")
|
||||
writeErr(w, &apierror.Error{Status: http.StatusBadGateway, Code: "destination_failed", Message: msg})
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, res)
|
||||
}
|
||||
|
||||
// POST /api/admin/backups/run — start a backup now. 202 once it has started;
|
||||
// 409 while one runs or before the destination is set up.
|
||||
func (h *handlers) handleRunBackup(w http.ResponseWriter, r *http.Request) {
|
||||
err := h.backups.Start(r.Context(), backup.TriggerManual)
|
||||
switch {
|
||||
case err == nil:
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
case errors.Is(err, backup.ErrRunning):
|
||||
writeErr(w, apierror.Conflict("backup_running", "A backup is already running."))
|
||||
case errors.Is(err, backup.ErrNotReady):
|
||||
writeErr(w, apierror.Conflict("backup_not_ready", "Set up where backups go first."))
|
||||
case errors.Is(err, backup.ErrClosed):
|
||||
writeErr(w, apierror.Conflict("shutting_down", "The server is shutting down."))
|
||||
default:
|
||||
writeErrWithLog(w, h.logger, "admin backups: start", apierror.Internal(err))
|
||||
}
|
||||
}
|
||||
+10
-1
@@ -15,6 +15,7 @@ import (
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/acquire"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/auth"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/backup"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/catalogue"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/config"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/coverart"
|
||||
@@ -40,7 +41,7 @@ import (
|
||||
// Mount attaches /api/* handlers to r. Public endpoints (login) are outside
|
||||
// RequireUser; everything else is gated by the middleware. The events writer
|
||||
// is shared with the Subsonic mount so /rest/scrobble feeds the same store.
|
||||
func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playevents.Writer, recCfg config.RecommendationConfig, recSettings *recsettings.Service, requestSvc *requests.Service, quarantineSvc *quarantine.Service, tracksSvc *tracks.Service, playlistsSvc *playlists.Service, coverEnricher *coverart.Enricher, coverSettings *coverart.SettingsService, tagSettings *tags.SettingsService, scanner *library.Scanner, scanCfg library.RunScanConfig, dataDir string, sender mailer.Sender, bus *eventbus.Bus, playlistScheduler *playlists.Scheduler, streamSecret []byte, netSettings *netsettings.Service, reacqSettings *reacquisition.SettingsService, fpSettings *library.FingerprintSettingsService, loudSettings *library.LoudnessSettingsService, acoustIDLookup *library.AcoustIDLookupWorker) {
|
||||
func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playevents.Writer, recCfg config.RecommendationConfig, recSettings *recsettings.Service, requestSvc *requests.Service, quarantineSvc *quarantine.Service, tracksSvc *tracks.Service, playlistsSvc *playlists.Service, coverEnricher *coverart.Enricher, coverSettings *coverart.SettingsService, tagSettings *tags.SettingsService, scanner *library.Scanner, scanCfg library.RunScanConfig, dataDir string, sender mailer.Sender, bus *eventbus.Bus, playlistScheduler *playlists.Scheduler, streamSecret []byte, netSettings *netsettings.Service, reacqSettings *reacquisition.SettingsService, fpSettings *library.FingerprintSettingsService, loudSettings *library.LoudnessSettingsService, acoustIDLookup *library.AcoustIDLookupWorker, backups *backup.Service) {
|
||||
rng := rand.New(rand.NewSource(rand.Int63()))
|
||||
setupToken, err := auth.NewSetupToken()
|
||||
if err != nil {
|
||||
@@ -74,6 +75,7 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playev
|
||||
fingerprintSettings: fpSettings,
|
||||
loudnessSettings: loudSettings,
|
||||
acoustIDLookup: acoustIDLookup,
|
||||
backups: backups,
|
||||
acquisition: acquire.NewStore(pool),
|
||||
catalogue: cat,
|
||||
wanted: wanted.New(pool, cat, logger),
|
||||
@@ -358,6 +360,11 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playev
|
||||
admin.Post("/tag-sources/{provider_id}/test", h.handleTestTagSource)
|
||||
admin.Post("/tag-sources/research", h.handleResearchTags)
|
||||
|
||||
admin.Get("/backups", h.handleGetBackups)
|
||||
admin.Put("/backups", h.handleUpdateBackups)
|
||||
admin.Post("/backups/test", h.handleTestBackups)
|
||||
admin.Post("/backups/run", h.handleRunBackup)
|
||||
|
||||
admin.Get("/smtp-config", h.handleGetSMTPConfig)
|
||||
admin.Put("/smtp-config", h.handleUpdateSMTPConfig)
|
||||
admin.Post("/smtp-config/test", h.handleTestSMTPConfig)
|
||||
@@ -429,6 +436,8 @@ type handlers struct {
|
||||
// acoustIDLookup is the AcoustID lookup worker: its settings, its
|
||||
// status and "look up now". Nil serves the defaults (off).
|
||||
acoustIDLookup *library.AcoustIDLookupWorker
|
||||
// backups is the server's own backup: settings, status, test and run.
|
||||
backups *backup.Service
|
||||
// acquisition is the configured indexers and download clients.
|
||||
acquisition *acquire.Store
|
||||
// catalogue is the release catalogue read from MusicBrainz, and
|
||||
|
||||
@@ -0,0 +1,711 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/secret"
|
||||
)
|
||||
|
||||
// Where a backup can go.
|
||||
const (
|
||||
DestinationFolder = "folder"
|
||||
DestinationS3 = "s3"
|
||||
)
|
||||
|
||||
// What started a run.
|
||||
const (
|
||||
TriggerSchedule = "schedule"
|
||||
TriggerManual = "manual"
|
||||
)
|
||||
|
||||
// The verdict a Status gives, worst first in the order an admin acts on them.
|
||||
const (
|
||||
StateOff = "off" // the nightly backup is switched off
|
||||
StateIncomplete = "incomplete" // switched on, but the destination isn't set up
|
||||
StateFailing = "failing" // the latest run failed
|
||||
StateWaiting = "waiting" // set up, and no backup has finished yet
|
||||
StateOverdue = "overdue" // the latest good backup is more than a day old
|
||||
StateOK = "ok"
|
||||
)
|
||||
|
||||
const (
|
||||
// RunLimit bounds one run, snapshot and upload together. The web panel
|
||||
// stops polling a run a little after this, since the run is over by then.
|
||||
RunLimit = 30 * time.Minute
|
||||
// A nightly backup that hasn't succeeded for this long is overdue; the
|
||||
// slack past a day covers a run that starts late or takes a while.
|
||||
overdueAfter = 26 * time.Hour
|
||||
tickEvery = 5 * time.Minute
|
||||
// A failed run is tried again after at least this long, and after longer
|
||||
// each time it fails again, so a destination that is down isn't hammered.
|
||||
retryAfter = time.Hour
|
||||
recentRuns = 10
|
||||
// Run history older than this is pruned after each run.
|
||||
keepRunsFor = 90 * 24 * time.Hour
|
||||
testTimeout = 30 * time.Second
|
||||
// The database backups' folder inside the destination. The music, when
|
||||
// it is backed up, sits beside it in the same destination.
|
||||
databaseDir = "database"
|
||||
)
|
||||
|
||||
var (
|
||||
// ErrRunning is returned by Start while a backup is running.
|
||||
ErrRunning = errors.New("backup: a backup is already running")
|
||||
// ErrNotReady is returned by Start when the destination isn't set up.
|
||||
ErrNotReady = errors.New("backup: the destination isn't set up yet")
|
||||
// ErrClosed is returned by Start once the server is shutting down.
|
||||
ErrClosed = errors.New("backup: the server is shutting down")
|
||||
)
|
||||
|
||||
// FieldError is a setting that can't be saved, with the reason in words.
|
||||
type FieldError struct {
|
||||
Field string `json:"field"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
func (e *FieldError) Error() string { return e.Field + ": " + e.Message }
|
||||
|
||||
// Config is the backup's settings as an admin sees them. The bucket's secret
|
||||
// key is never sent back, only whether one is stored.
|
||||
type Config struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
Destination string `json:"destination"`
|
||||
Folder string `json:"folder"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
Bucket string `json:"bucket"`
|
||||
AccessKey string `json:"access_key"`
|
||||
HasSecretKey bool `json:"has_secret_key"`
|
||||
RunHour int `json:"run_hour"`
|
||||
UploadLimitKiB int `json:"upload_limit_kib"`
|
||||
}
|
||||
|
||||
// Update is a save from the admin card. An empty SecretKey keeps the stored
|
||||
// one, so the key doesn't have to be typed again on every save.
|
||||
type Update struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
Destination string `json:"destination"`
|
||||
Folder string `json:"folder"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
Bucket string `json:"bucket"`
|
||||
AccessKey string `json:"access_key"`
|
||||
SecretKey string `json:"secret_key"`
|
||||
RunHour int `json:"run_hour"`
|
||||
UploadLimitKiB int `json:"upload_limit_kib"`
|
||||
}
|
||||
|
||||
// Run is one attempt at a backup.
|
||||
type Run struct {
|
||||
ID string `json:"id"`
|
||||
Trigger string `json:"trigger"`
|
||||
Status string `json:"status"`
|
||||
StartedAt time.Time `json:"started_at"`
|
||||
FinishedAt *time.Time `json:"finished_at"`
|
||||
Location string `json:"location"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
|
||||
// Status is everything the admin panel shows: a verdict, what stands in its
|
||||
// way, and the recent runs.
|
||||
type Status struct {
|
||||
Config Config `json:"config"`
|
||||
State string `json:"state"`
|
||||
Problems []string `json:"problems"`
|
||||
RunningSince *time.Time `json:"running_since"`
|
||||
LastSuccess *time.Time `json:"last_success"`
|
||||
FailingSince *time.Time `json:"failing_since"`
|
||||
// NextRun is when the schedule starts the next backup, or nil when it
|
||||
// won't: switched off, or not set up.
|
||||
NextRun *time.Time `json:"next_run"`
|
||||
RunLimitSeconds int `json:"run_limit_seconds"`
|
||||
Runs []Run `json:"runs"`
|
||||
}
|
||||
|
||||
// TestResult is what a destination test found: how many backups are there
|
||||
// already and when the newest was made.
|
||||
type TestResult struct {
|
||||
Backups int `json:"backups"`
|
||||
Latest *time.Time `json:"latest"`
|
||||
}
|
||||
|
||||
// Service runs Minstrel's own backup: nightly when switched on, or when an
|
||||
// admin asks. One run at a time; each is recorded in backup_runs.
|
||||
type Service struct {
|
||||
pool *pgxpool.Pool
|
||||
logger *slog.Logger
|
||||
build string
|
||||
now func() time.Time
|
||||
|
||||
base context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
|
||||
mu sync.Mutex
|
||||
runningSince time.Time
|
||||
closed bool
|
||||
}
|
||||
|
||||
// NewService returns a Service over pool. build is the server's version,
|
||||
// written into each backup's manifest.
|
||||
func NewService(pool *pgxpool.Pool, logger *slog.Logger, build string) *Service {
|
||||
base, cancel := context.WithCancel(context.Background())
|
||||
return &Service{pool: pool, logger: logger, build: build, now: time.Now, base: base, cancel: cancel}
|
||||
}
|
||||
|
||||
func (s *Service) settings(ctx context.Context) (dbq.BackupSetting, error) {
|
||||
st, err := dbq.New(s.pool).GetBackupSettings(ctx)
|
||||
if err != nil {
|
||||
return st, fmt.Errorf("backup: read settings: %w", err)
|
||||
}
|
||||
return st, nil
|
||||
}
|
||||
|
||||
func view(st dbq.BackupSetting) Config {
|
||||
return Config{
|
||||
Enabled: st.Enabled,
|
||||
Destination: st.Destination,
|
||||
Folder: st.Folder,
|
||||
Endpoint: st.Endpoint,
|
||||
Bucket: st.Bucket,
|
||||
AccessKey: st.AccessKey,
|
||||
HasSecretKey: st.SecretKey != "",
|
||||
RunHour: int(st.RunHour),
|
||||
UploadLimitKiB: int(st.UploadLimitKib),
|
||||
}
|
||||
}
|
||||
|
||||
// Config returns the settings as an admin sees them.
|
||||
func (s *Service) Config(ctx context.Context) (Config, error) {
|
||||
st, err := s.settings(ctx)
|
||||
return view(st), err
|
||||
}
|
||||
|
||||
// merge applies u over the stored settings: it checks each field, and seals
|
||||
// a new secret key. It returns a *FieldError for a value that can't be kept.
|
||||
func merge(st dbq.BackupSetting, u Update) (dbq.BackupSetting, error) {
|
||||
if u.Destination != DestinationFolder && u.Destination != DestinationS3 {
|
||||
return st, &FieldError{"destination", "Choose a folder or an S3 bucket."}
|
||||
}
|
||||
if u.RunHour < 0 || u.RunHour > 23 {
|
||||
return st, &FieldError{"run_hour", "The hour is from 0 to 23."}
|
||||
}
|
||||
if u.UploadLimitKiB < 0 || u.UploadLimitKiB > 1<<30 {
|
||||
return st, &FieldError{"upload_limit_kib", "The upload limit is 0 (no limit) or more."}
|
||||
}
|
||||
folder := strings.TrimSpace(u.Folder)
|
||||
if folder != "" && !filepath.IsAbs(folder) {
|
||||
return st, &FieldError{"folder", "Give the folder's full path, starting with /."}
|
||||
}
|
||||
endpoint := strings.TrimRight(strings.TrimSpace(u.Endpoint), "/")
|
||||
if endpoint != "" {
|
||||
e, err := url.Parse(endpoint)
|
||||
if err != nil || (e.Scheme != "http" && e.Scheme != "https") || e.Host == "" {
|
||||
return st, &FieldError{"endpoint", "Give the endpoint as a URL, such as https://s3.example.com."}
|
||||
}
|
||||
}
|
||||
st.Enabled = u.Enabled
|
||||
st.Destination = u.Destination
|
||||
st.Folder = filepath.Clean(folder)
|
||||
if folder == "" {
|
||||
st.Folder = ""
|
||||
}
|
||||
st.Endpoint = endpoint
|
||||
st.Bucket = strings.TrimSpace(u.Bucket)
|
||||
st.AccessKey = strings.TrimSpace(u.AccessKey)
|
||||
st.RunHour = int16(u.RunHour)
|
||||
st.UploadLimitKib = int32(u.UploadLimitKiB)
|
||||
if u.SecretKey != "" {
|
||||
sealed, err := secret.Seal(secret.BackupSecretKey.Purpose(), "", u.SecretKey)
|
||||
if err != nil {
|
||||
return st, err
|
||||
}
|
||||
st.SecretKey = sealed
|
||||
}
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// Save stores u. A value that can't be kept is a *FieldError and nothing is
|
||||
// stored. Settings that are incomplete still save; Status lists what's missing.
|
||||
func (s *Service) Save(ctx context.Context, u Update) error {
|
||||
st, err := s.settings(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
st, err = merge(st, u)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return dbq.New(s.pool).SaveBackupSettings(ctx, dbq.SaveBackupSettingsParams{
|
||||
Enabled: st.Enabled,
|
||||
Destination: st.Destination,
|
||||
Folder: st.Folder,
|
||||
Endpoint: st.Endpoint,
|
||||
Bucket: st.Bucket,
|
||||
AccessKey: st.AccessKey,
|
||||
SecretKey: st.SecretKey,
|
||||
RunHour: st.RunHour,
|
||||
UploadLimitKib: st.UploadLimitKib,
|
||||
})
|
||||
}
|
||||
|
||||
// problems is what stops a backup from running with these settings, in words.
|
||||
func problems(st dbq.BackupSetting) []string {
|
||||
var out []string
|
||||
switch st.Destination {
|
||||
case DestinationFolder:
|
||||
if st.Folder == "" {
|
||||
out = append(out, "Choose the folder backups are written to.")
|
||||
}
|
||||
case DestinationS3:
|
||||
if st.Endpoint == "" {
|
||||
out = append(out, "Enter the bucket's endpoint.")
|
||||
}
|
||||
if st.Bucket == "" {
|
||||
out = append(out, "Enter the bucket's name.")
|
||||
}
|
||||
if st.AccessKey == "" {
|
||||
out = append(out, "Enter the access key.")
|
||||
}
|
||||
if st.SecretKey == "" {
|
||||
out = append(out, "Enter the secret key.")
|
||||
}
|
||||
default:
|
||||
out = append(out, "Choose where backups go.")
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func timeOf(t pgtype.Timestamptz) time.Time {
|
||||
if !t.Valid || t.Time.Unix() <= 0 {
|
||||
return time.Time{}
|
||||
}
|
||||
return t.Time
|
||||
}
|
||||
|
||||
func ptr(t time.Time) *time.Time {
|
||||
if t.IsZero() {
|
||||
return nil
|
||||
}
|
||||
return &t
|
||||
}
|
||||
|
||||
// runTimes is what the verdict and the schedule turn on. A zero time is a run
|
||||
// that never happened.
|
||||
type runTimes struct {
|
||||
lastSuccess, lastAttempt, failingSince time.Time
|
||||
}
|
||||
|
||||
func (s *Service) times(ctx context.Context) (runTimes, error) {
|
||||
row, err := dbq.New(s.pool).BackupRunTimes(ctx)
|
||||
if err != nil {
|
||||
return runTimes{}, fmt.Errorf("backup: read run times: %w", err)
|
||||
}
|
||||
return runTimes{timeOf(row.LastSuccess), timeOf(row.LastAttempt), timeOf(row.FailingSince)}, nil
|
||||
}
|
||||
|
||||
// slotBefore is the latest scheduled start at or before now: today at hour
|
||||
// UTC, or yesterday's if that hour hasn't come yet.
|
||||
func slotBefore(now time.Time, hour int) time.Time {
|
||||
now = now.UTC()
|
||||
slot := time.Date(now.Year(), now.Month(), now.Day(), hour, 0, 0, 0, time.UTC)
|
||||
if slot.After(now) {
|
||||
slot = slot.AddDate(0, 0, -1)
|
||||
}
|
||||
return slot
|
||||
}
|
||||
|
||||
// due reports when the schedule next starts a backup: now or earlier means
|
||||
// the next tick starts one. A slot that passed with no attempt since is due,
|
||||
// so a server that was down at the hour catches up. A run that failed is
|
||||
// tried again, waiting at least retryAfter and then as long again as it has
|
||||
// been failing, until it succeeds or the next slot comes.
|
||||
func due(now time.Time, hour int, t runTimes) time.Time {
|
||||
slot := slotBefore(now, hour)
|
||||
if t.lastAttempt.Before(slot) {
|
||||
return slot
|
||||
}
|
||||
next := slot.AddDate(0, 0, 1)
|
||||
if !t.failingSince.IsZero() {
|
||||
wait := max(retryAfter, t.lastAttempt.Sub(t.failingSince))
|
||||
if retry := t.lastAttempt.Add(wait); retry.Before(next) {
|
||||
return retry
|
||||
}
|
||||
}
|
||||
return next
|
||||
}
|
||||
|
||||
func verdict(st dbq.BackupSetting, probs []string, t runTimes, now time.Time) string {
|
||||
switch {
|
||||
case !st.Enabled:
|
||||
return StateOff
|
||||
case len(probs) > 0:
|
||||
return StateIncomplete
|
||||
case !t.failingSince.IsZero():
|
||||
return StateFailing
|
||||
case t.lastSuccess.IsZero():
|
||||
return StateWaiting
|
||||
case now.Sub(t.lastSuccess) > overdueAfter:
|
||||
return StateOverdue
|
||||
}
|
||||
return StateOK
|
||||
}
|
||||
|
||||
func uuidText(u pgtype.UUID) string {
|
||||
b := u.Bytes
|
||||
return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16])
|
||||
}
|
||||
|
||||
// Status is the verdict, what stands in its way, and the recent runs.
|
||||
func (s *Service) Status(ctx context.Context) (Status, error) {
|
||||
st, err := s.settings(ctx)
|
||||
if err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
t, err := s.times(ctx)
|
||||
if err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
rows, err := dbq.New(s.pool).ListBackupRuns(ctx, recentRuns)
|
||||
if err != nil {
|
||||
return Status{}, fmt.Errorf("backup: list runs: %w", err)
|
||||
}
|
||||
now := s.now()
|
||||
probs := problems(st)
|
||||
out := Status{
|
||||
Config: view(st),
|
||||
State: verdict(st, probs, t, now),
|
||||
Problems: probs,
|
||||
LastSuccess: ptr(t.lastSuccess),
|
||||
FailingSince: ptr(t.failingSince),
|
||||
RunLimitSeconds: int(RunLimit / time.Second),
|
||||
Runs: make([]Run, 0, len(rows)),
|
||||
}
|
||||
if out.Problems == nil {
|
||||
out.Problems = []string{}
|
||||
}
|
||||
s.mu.Lock()
|
||||
out.RunningSince = ptr(s.runningSince)
|
||||
s.mu.Unlock()
|
||||
if st.Enabled && len(probs) == 0 {
|
||||
next := due(now, int(st.RunHour), t)
|
||||
if next.Before(now) {
|
||||
// Due now: the next tick starts it.
|
||||
next = now
|
||||
}
|
||||
out.NextRun = &next
|
||||
}
|
||||
for _, r := range rows {
|
||||
out.Runs = append(out.Runs, Run{
|
||||
ID: uuidText(r.ID),
|
||||
Trigger: r.Trigger,
|
||||
Status: r.Status,
|
||||
StartedAt: r.StartedAt.Time,
|
||||
FinishedAt: ptr(timeOf(r.FinishedAt)),
|
||||
Location: r.Location,
|
||||
Bytes: r.Bytes,
|
||||
Error: r.Error,
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func bucketOf(st dbq.BackupSetting) (Bucket, error) {
|
||||
key, err := secret.Open(secret.BackupSecretKey.Purpose(), "", st.SecretKey)
|
||||
if err != nil {
|
||||
return Bucket{}, fmt.Errorf("backup: the stored secret key doesn't open; enter it again: %w", err)
|
||||
}
|
||||
return Bucket{Endpoint: st.Endpoint, Name: st.Bucket, AccessKey: st.AccessKey, SecretKey: key}, nil
|
||||
}
|
||||
|
||||
// Test checks the destination u describes, falling back to the stored
|
||||
// secret key when u leaves it empty. It only reads: it lists the backups
|
||||
// already there and writes nothing.
|
||||
func (s *Service) Test(ctx context.Context, u Update) (TestResult, error) {
|
||||
st, err := s.settings(ctx)
|
||||
if err != nil {
|
||||
return TestResult{}, err
|
||||
}
|
||||
st, err = merge(st, u)
|
||||
if err != nil {
|
||||
return TestResult{}, err
|
||||
}
|
||||
if probs := problems(st); len(probs) > 0 {
|
||||
return TestResult{}, fmt.Errorf("%w: %s", ErrNotReady, strings.Join(probs, " "))
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, testTimeout)
|
||||
defer cancel()
|
||||
|
||||
var res TestResult
|
||||
note := func(name string, at time.Time) {
|
||||
if !IsFileName(name) {
|
||||
return
|
||||
}
|
||||
res.Backups++
|
||||
if res.Latest == nil || at.After(*res.Latest) {
|
||||
res.Latest = &at
|
||||
}
|
||||
}
|
||||
if st.Destination == DestinationFolder {
|
||||
if err := isFolder(st.Folder); err != nil {
|
||||
return res, err
|
||||
}
|
||||
entries, err := os.ReadDir(filepath.Join(st.Folder, databaseDir))
|
||||
if err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return res, fmt.Errorf("backup: read %s: %w", filepath.Join(st.Folder, databaseDir), err)
|
||||
}
|
||||
for _, e := range entries {
|
||||
if info, err := e.Info(); err == nil && info.Mode().IsRegular() {
|
||||
note(e.Name(), info.ModTime())
|
||||
}
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
b, err := bucketOf(st)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
objs, err := b.List(ctx, databaseDir+"/", 1000)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
for _, o := range objs {
|
||||
note(path.Base(o.Key), o.LastModified)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// isFolder refuses a folder that isn't there. A backup never creates the
|
||||
// destination folder itself: a missing one usually means a volume that
|
||||
// isn't mounted, and writing beside it would fill the server's own disk.
|
||||
func isFolder(dir string) error {
|
||||
info, err := os.Stat(dir)
|
||||
if err != nil {
|
||||
return fmt.Errorf("backup: the folder %s isn't there; is its volume mounted? %w", dir, err)
|
||||
}
|
||||
if !info.IsDir() {
|
||||
return fmt.Errorf("backup: %s is a file, not a folder", dir)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Start begins a backup in the background and returns at once. It returns
|
||||
// ErrRunning while one runs, ErrNotReady when the destination isn't set up,
|
||||
// and ErrClosed once the server is shutting down.
|
||||
func (s *Service) Start(ctx context.Context, trigger string) error {
|
||||
st, err := s.settings(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(problems(st)) > 0 {
|
||||
return ErrNotReady
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.closed {
|
||||
return ErrClosed
|
||||
}
|
||||
if !s.runningSince.IsZero() {
|
||||
return ErrRunning
|
||||
}
|
||||
s.runningSince = s.now()
|
||||
s.wg.Add(1)
|
||||
go func() {
|
||||
defer s.wg.Done()
|
||||
defer func() {
|
||||
s.mu.Lock()
|
||||
s.runningSince = time.Time{}
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
s.run(trigger, st)
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Running reports when the running backup started, or the zero time.
|
||||
func (s *Service) Running() time.Time {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.runningSince
|
||||
}
|
||||
|
||||
func (s *Service) run(trigger string, st dbq.BackupSetting) {
|
||||
ctx, cancel := context.WithTimeout(s.base, RunLimit)
|
||||
defer cancel()
|
||||
q := dbq.New(s.pool)
|
||||
started, err := q.StartBackupRun(ctx, trigger)
|
||||
if err != nil {
|
||||
s.logger.Error("backup: record the run's start", "err", err)
|
||||
return
|
||||
}
|
||||
location, size, err := s.write(ctx, st, started.StartedAt.Time)
|
||||
|
||||
// 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}
|
||||
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)
|
||||
}
|
||||
if err := q.FinishBackupRun(fctx, fin); err != nil {
|
||||
s.logger.Error("backup: record the run's end", "err", err)
|
||||
}
|
||||
if _, err := q.PruneBackupRuns(fctx, pgtype.Timestamptz{Time: s.now().Add(-keepRunsFor), Valid: true}); err != nil {
|
||||
s.logger.Warn("backup: prune old runs", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// describe is a run's error as the panel shows it.
|
||||
func describe(err error) string {
|
||||
switch {
|
||||
case errors.Is(err, context.DeadlineExceeded):
|
||||
return fmt.Sprintf("The backup took longer than %d minutes and was stopped.", int(RunLimit/time.Minute))
|
||||
case errors.Is(err, context.Canceled):
|
||||
return "The server stopped while the backup was running."
|
||||
}
|
||||
return strings.TrimPrefix(err.Error(), "backup: ")
|
||||
}
|
||||
|
||||
// write snapshots the database into one backup file at the destination and
|
||||
// returns where it went and its size.
|
||||
func (s *Service) write(ctx context.Context, st dbq.BackupSetting, at time.Time) (string, int64, error) {
|
||||
if st.Destination == DestinationFolder {
|
||||
// Checked before the snapshot, so a missing volume costs nothing.
|
||||
if err := isFolder(st.Folder); err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
}
|
||||
c, err := Snapshot(ctx, s.pool)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
if st.Destination == DestinationFolder {
|
||||
dir := filepath.Join(st.Folder, databaseDir)
|
||||
if err := os.Mkdir(dir, 0o700); err != nil && !errors.Is(err, os.ErrExist) {
|
||||
return "", 0, fmt.Errorf("backup: create %s: %w", dir, err)
|
||||
}
|
||||
file, _, err := WriteFile(dir, c, at, s.build)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
info, err := os.Stat(file)
|
||||
if err != nil {
|
||||
return file, 0, err
|
||||
}
|
||||
return file, info.Size(), nil
|
||||
}
|
||||
|
||||
b, err := bucketOf(st)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
// The file is written to a temporary file first: S3 signs over the
|
||||
// body's hash and needs its length before the upload starts.
|
||||
tmp, err := os.CreateTemp("", ".minstrel-backup-*.tar")
|
||||
if err != nil {
|
||||
return "", 0, fmt.Errorf("backup: create a temporary file: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
_ = tmp.Close()
|
||||
_ = os.Remove(tmp.Name())
|
||||
}()
|
||||
h := sha256.New()
|
||||
if _, err := Write(io.MultiWriter(tmp, h), c, at, s.build); err != nil {
|
||||
return "", 0, fmt.Errorf("backup: write the temporary file: %w", err)
|
||||
}
|
||||
size, err := tmp.Seek(0, io.SeekCurrent)
|
||||
if err == nil {
|
||||
_, err = tmp.Seek(0, io.SeekStart)
|
||||
}
|
||||
if err != nil {
|
||||
return "", 0, fmt.Errorf("backup: rewind the temporary file: %w", err)
|
||||
}
|
||||
key := databaseDir + "/" + FileName(at)
|
||||
body := Throttle(ctx, tmp, int64(st.UploadLimitKib)*1024)
|
||||
if err := b.Put(ctx, key, body, size, hex.EncodeToString(h.Sum(nil))); err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
return key, size, nil
|
||||
}
|
||||
|
||||
// Tick starts a scheduled backup when one is due. A backup that is already
|
||||
// running, or settings that aren't ready, mean there's nothing to do.
|
||||
func (s *Service) Tick(ctx context.Context) {
|
||||
st, err := s.settings(ctx)
|
||||
if err != nil {
|
||||
s.logger.Warn("backup: tick", "err", err)
|
||||
return
|
||||
}
|
||||
if !st.Enabled || len(problems(st)) > 0 {
|
||||
return
|
||||
}
|
||||
t, err := s.times(ctx)
|
||||
if err != nil {
|
||||
s.logger.Warn("backup: tick", "err", err)
|
||||
return
|
||||
}
|
||||
if due(s.now(), int(st.RunHour), t).After(s.now()) {
|
||||
return
|
||||
}
|
||||
if err := s.Start(ctx, TriggerSchedule); err != nil && !errors.Is(err, ErrRunning) && !errors.Is(err, ErrClosed) {
|
||||
s.logger.Warn("backup: start the scheduled backup", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Recover marks runs a stopped server left running as failed. It runs once
|
||||
// at startup, before the first tick.
|
||||
func (s *Service) Recover(ctx context.Context) {
|
||||
n, err := dbq.New(s.pool).FailUnfinishedBackupRuns(ctx)
|
||||
if err != nil {
|
||||
s.logger.Warn("backup: recover unfinished runs", "err", err)
|
||||
return
|
||||
}
|
||||
if n > 0 {
|
||||
s.logger.Warn("backup: a run was cut off by the last shutdown", "runs", n)
|
||||
}
|
||||
}
|
||||
|
||||
// Loop recovers, then ticks until ctx ends. The first tick waits a full
|
||||
// interval, so a restart isn't followed straight away by a backup.
|
||||
func (s *Service) Loop(ctx context.Context) {
|
||||
s.Recover(ctx)
|
||||
t := time.NewTicker(tickEvery)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
s.Tick(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Close stops a running backup and waits for it to record how it ended.
|
||||
func (s *Service) Close() {
|
||||
s.mu.Lock()
|
||||
s.closed = true
|
||||
s.mu.Unlock()
|
||||
s.cancel()
|
||||
s.wg.Wait()
|
||||
}
|
||||
@@ -0,0 +1,281 @@
|
||||
package backup_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/backup"
|
||||
)
|
||||
|
||||
func newService(t *testing.T) (*backup.Service, *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
pool := newPool(t)
|
||||
svc := backup.NewService(pool, slog.New(slog.NewTextHandler(io.Discard, nil)), "test-build")
|
||||
t.Cleanup(svc.Close)
|
||||
return svc, pool
|
||||
}
|
||||
|
||||
// waitIdle waits for the running backup to finish, for at most 30 seconds.
|
||||
func waitIdle(t *testing.T, svc *backup.Service) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(30 * time.Second)
|
||||
for !svc.Running().IsZero() {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("the backup is still running after 30s")
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func status(t *testing.T, svc *backup.Service) backup.Status {
|
||||
t.Helper()
|
||||
st, err := svc.Status(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
func TestABackupIsOffAndRefusesToStartUntilSetUp_Integration(t *testing.T) {
|
||||
svc, _ := newService(t)
|
||||
st := status(t, svc)
|
||||
if st.State != backup.StateOff || st.Config.Enabled || st.NextRun != nil || len(st.Runs) != 0 {
|
||||
t.Errorf("a fresh server's backup = %+v, want off with no runs", st)
|
||||
}
|
||||
if err := svc.Start(context.Background(), backup.TriggerManual); !errors.Is(err, backup.ErrNotReady) {
|
||||
t.Errorf("Start = %v, want ErrNotReady", err)
|
||||
}
|
||||
if err := svc.Save(context.Background(), backup.Update{Enabled: true, Destination: backup.DestinationS3, RunHour: 3}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st := status(t, svc); st.State != backup.StateIncomplete || len(st.Problems) != 4 {
|
||||
t.Errorf("switched on with no bucket: %s %v", st.State, st.Problems)
|
||||
}
|
||||
}
|
||||
|
||||
func TestABackupRunsIntoAFolder_Integration(t *testing.T) {
|
||||
svc, _ := newService(t)
|
||||
dir := t.TempDir()
|
||||
ctx := context.Background()
|
||||
if err := svc.Save(ctx, backup.Update{Enabled: true, Destination: backup.DestinationFolder, Folder: dir, RunHour: 3}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st := status(t, svc); st.State != backup.StateWaiting || st.NextRun == nil {
|
||||
t.Errorf("set up, never run: %s next %v", st.State, st.NextRun)
|
||||
}
|
||||
if err := svc.Start(ctx, backup.TriggerManual); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitIdle(t, svc)
|
||||
|
||||
st := status(t, svc)
|
||||
if st.State != backup.StateOK || len(st.Runs) != 1 || st.LastSuccess == nil {
|
||||
t.Fatalf("after a run: %+v", st)
|
||||
}
|
||||
run := st.Runs[0]
|
||||
if run.Status != "succeeded" || run.Trigger != backup.TriggerManual || run.FinishedAt == nil ||
|
||||
filepath.Dir(run.Location) != filepath.Join(dir, "database") || run.Error != "" {
|
||||
t.Fatalf("run = %+v", run)
|
||||
}
|
||||
f, err := os.Open(run.Location)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _ = f.Close() }()
|
||||
info, _ := f.Stat()
|
||||
if info.Size() != run.Bytes {
|
||||
t.Errorf("recorded %d bytes, the file holds %d", run.Bytes, info.Size())
|
||||
}
|
||||
m, _, err := backup.Read(f)
|
||||
if err != nil || m.Build != "test-build" {
|
||||
t.Errorf("the backup reads back as %+v, %v", m, err)
|
||||
}
|
||||
|
||||
res, err := svc.Test(ctx, backup.Update{Destination: backup.DestinationFolder, Folder: dir, RunHour: 3})
|
||||
if err != nil || res.Backups != 1 || res.Latest == nil {
|
||||
t.Errorf("Test = %+v, %v; want the one backup", res, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMissingFolderFailsTheRunAndSaysSo_Integration(t *testing.T) {
|
||||
svc, _ := newService(t)
|
||||
ctx := context.Background()
|
||||
missing := filepath.Join(t.TempDir(), "not-mounted")
|
||||
if err := svc.Save(ctx, backup.Update{Enabled: true, Destination: backup.DestinationFolder, Folder: missing, RunHour: 3}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := svc.Test(ctx, backup.Update{Destination: backup.DestinationFolder, Folder: missing}); err == nil {
|
||||
t.Error("Test passed on a missing folder")
|
||||
}
|
||||
if err := svc.Start(ctx, backup.TriggerManual); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitIdle(t, svc)
|
||||
st := status(t, svc)
|
||||
if st.State != backup.StateFailing || st.FailingSince == nil || len(st.Runs) != 1 ||
|
||||
!strings.Contains(st.Runs[0].Error, "isn't there") {
|
||||
t.Errorf("after a failed run: %s %+v", st.State, st.Runs)
|
||||
}
|
||||
if _, err := os.Stat(missing); !os.IsNotExist(err) {
|
||||
t.Error("the run created the missing folder")
|
||||
}
|
||||
}
|
||||
|
||||
// fakeBucket answers PUTs and lists like an S3 bucket. While hold is open a
|
||||
// PUT waits on it.
|
||||
type fakeBucket struct {
|
||||
mu sync.Mutex
|
||||
puts map[string][]byte
|
||||
hold chan struct{}
|
||||
}
|
||||
|
||||
func (b *fakeBucket) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if !strings.HasPrefix(r.Header.Get("Authorization"), "AWS4-HMAC-SHA256 Credential=ak/") {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
switch r.Method {
|
||||
case http.MethodPut:
|
||||
if b.hold != nil {
|
||||
<-b.hold
|
||||
}
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
b.mu.Lock()
|
||||
b.puts[r.URL.Path] = body
|
||||
b.mu.Unlock()
|
||||
case http.MethodGet:
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
_, _ = fmt.Fprint(w, `<ListBucketResult><IsTruncated>false</IsTruncated>`)
|
||||
for p, body := range b.puts {
|
||||
_, _ = fmt.Fprintf(w, `<Contents><Key>%s</Key><Size>%d</Size><LastModified>2026-11-02T03:00:05.000Z</LastModified></Contents>`,
|
||||
strings.TrimPrefix(p, "/minstrel/"), len(body))
|
||||
}
|
||||
_, _ = fmt.Fprint(w, `</ListBucketResult>`)
|
||||
}
|
||||
}
|
||||
|
||||
func bucketUpdate(endpoint string) backup.Update {
|
||||
return backup.Update{Enabled: true, Destination: backup.DestinationS3, Endpoint: endpoint,
|
||||
Bucket: "minstrel", AccessKey: "ak", SecretKey: "sk", RunHour: 3}
|
||||
}
|
||||
|
||||
func TestABackupRunsIntoABucket_Integration(t *testing.T) {
|
||||
svc, pool := newService(t)
|
||||
fake := &fakeBucket{puts: map[string][]byte{}}
|
||||
srv := httptest.NewServer(fake)
|
||||
defer srv.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
if err := svc.Save(ctx, bucketUpdate(srv.URL)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var stored string
|
||||
if err := pool.QueryRow(ctx, "SELECT secret_key FROM backup_settings").Scan(&stored); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.HasPrefix(stored, "v1.") {
|
||||
t.Fatalf("the secret key is stored as %q, want it sealed", stored)
|
||||
}
|
||||
if cfg, _ := svc.Config(ctx); !cfg.HasSecretKey {
|
||||
t.Error("the config doesn't say a secret key is stored")
|
||||
}
|
||||
if err := svc.Start(ctx, backup.TriggerManual); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitIdle(t, svc)
|
||||
|
||||
st := status(t, svc)
|
||||
if st.State != backup.StateOK || len(st.Runs) != 1 || st.Runs[0].Status != "succeeded" {
|
||||
t.Fatalf("after a run: %s %+v", st.State, st.Runs)
|
||||
}
|
||||
key := st.Runs[0].Location
|
||||
if !strings.HasPrefix(key, "database/minstrel-") || !backup.IsFileName(filepath.Base(key)) {
|
||||
t.Errorf("stored under %q", key)
|
||||
}
|
||||
fake.mu.Lock()
|
||||
body := fake.puts["/minstrel/"+key]
|
||||
fake.mu.Unlock()
|
||||
if int64(len(body)) != st.Runs[0].Bytes {
|
||||
t.Errorf("the bucket got %d bytes, the run recorded %d", len(body), st.Runs[0].Bytes)
|
||||
}
|
||||
if _, _, err := backup.Read(bytes.NewReader(body)); err != nil {
|
||||
t.Errorf("the uploaded backup doesn't read: %v", err)
|
||||
}
|
||||
|
||||
// The test sends no secret key: it uses the stored one, and only lists.
|
||||
u := bucketUpdate(srv.URL)
|
||||
u.SecretKey = ""
|
||||
res, err := svc.Test(ctx, u)
|
||||
if err != nil || res.Backups != 1 {
|
||||
t.Errorf("Test = %+v, %v; want the one backup", res, err)
|
||||
}
|
||||
fake.mu.Lock()
|
||||
defer fake.mu.Unlock()
|
||||
if len(fake.puts) != 1 {
|
||||
t.Errorf("the test wrote to the bucket: %d objects", len(fake.puts))
|
||||
}
|
||||
}
|
||||
|
||||
func TestOneBackupAtATime_Integration(t *testing.T) {
|
||||
svc, _ := newService(t)
|
||||
fake := &fakeBucket{puts: map[string][]byte{}, hold: make(chan struct{})}
|
||||
srv := httptest.NewServer(fake)
|
||||
defer srv.Close()
|
||||
ctx := context.Background()
|
||||
if err := svc.Save(ctx, bucketUpdate(srv.URL)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := svc.Start(ctx, backup.TriggerManual); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := svc.Start(ctx, backup.TriggerManual); !errors.Is(err, backup.ErrRunning) {
|
||||
t.Errorf("a second Start = %v, want ErrRunning", err)
|
||||
}
|
||||
if st := status(t, svc); st.RunningSince == nil {
|
||||
t.Error("the status doesn't say a backup is running")
|
||||
}
|
||||
close(fake.hold)
|
||||
waitIdle(t, svc)
|
||||
}
|
||||
|
||||
func TestRecoverFailsARunTheLastShutdownCutOff_Integration(t *testing.T) {
|
||||
svc, pool := newService(t)
|
||||
ctx := context.Background()
|
||||
if _, err := pool.Exec(ctx, "INSERT INTO backup_runs (trigger) VALUES ('schedule')"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc.Recover(ctx)
|
||||
st := status(t, svc)
|
||||
if len(st.Runs) != 1 || st.Runs[0].Status != "failed" || st.Runs[0].FinishedAt == nil ||
|
||||
!strings.Contains(st.Runs[0].Error, "server stopped") {
|
||||
t.Errorf("runs = %+v", st.Runs)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSaveRefusesABadValueAndKeepsTheOldSettings_Integration(t *testing.T) {
|
||||
svc, _ := newService(t)
|
||||
ctx := context.Background()
|
||||
err := svc.Save(ctx, backup.Update{Enabled: true, Destination: backup.DestinationFolder, Folder: "relative/path"})
|
||||
var fe *backup.FieldError
|
||||
if !errors.As(err, &fe) || fe.Field != "folder" {
|
||||
t.Fatalf("err = %v, want a FieldError on folder", err)
|
||||
}
|
||||
if cfg, _ := svc.Config(ctx); cfg.Enabled || cfg.Folder != "" {
|
||||
t.Errorf("a refused save changed the settings: %+v", cfg)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
func at(s string) time.Time {
|
||||
t, err := time.Parse(time.RFC3339, s)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
func TestDueRunsAtTheHourCatchesUpAndBacksOffAfterAFailure(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
now string
|
||||
t runTimes
|
||||
want string
|
||||
}{
|
||||
{"never run: the slot that passed is due", "2026-11-02T10:00:00Z",
|
||||
runTimes{}, "2026-11-02T03:00:00Z"},
|
||||
{"ran at the hour: tomorrow's slot", "2026-11-02T10:00:00Z",
|
||||
runTimes{lastSuccess: at("2026-11-02T03:00:05Z"), lastAttempt: at("2026-11-02T03:00:05Z")}, "2026-11-03T03:00:00Z"},
|
||||
{"before today's hour, yesterday's ran: today's slot", "2026-11-02T02:00:00Z",
|
||||
runTimes{lastSuccess: at("2026-11-01T03:00:05Z"), lastAttempt: at("2026-11-01T03:00:05Z")}, "2026-11-02T03:00:00Z"},
|
||||
{"server was down at the hour: catches up", "2026-11-02T10:00:00Z",
|
||||
runTimes{lastSuccess: at("2026-11-01T03:00:05Z"), lastAttempt: at("2026-11-01T03:00:05Z")}, "2026-11-02T03:00:00Z"},
|
||||
{"first failure: an hour later", "2026-11-02T03:30:00Z",
|
||||
runTimes{lastAttempt: at("2026-11-02T03:00:00Z"), failingSince: at("2026-11-02T03:00:00Z")}, "2026-11-02T04:00:00Z"},
|
||||
{"failing for two hours: two hours later", "2026-11-02T05:30:00Z",
|
||||
runTimes{lastAttempt: at("2026-11-02T05:00:00Z"), failingSince: at("2026-11-02T03:00:00Z")}, "2026-11-02T07:00:00Z"},
|
||||
{"a retry past the next slot waits for the slot", "2026-11-02T19:30:00Z",
|
||||
runTimes{lastAttempt: at("2026-11-02T19:00:00Z"), failingSince: at("2026-11-02T03:00:00Z")}, "2026-11-03T03:00:00Z"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := due(at(c.now), 3, c.t); !got.Equal(at(c.want)) {
|
||||
t.Errorf("%s: due = %s, want %s", c.name, got.Format(time.RFC3339), c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheHourIsUTC(t *testing.T) {
|
||||
// 22:00 at UTC-5 is 03:00 UTC the next day, which is that day's slot.
|
||||
now := time.Date(2026, 11, 1, 22, 0, 0, 0, time.FixedZone("x", -5*3600))
|
||||
if got := slotBefore(now, 3); !got.Equal(at("2026-11-02T03:00:00Z")) {
|
||||
t.Errorf("slot = %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerdict(t *testing.T) {
|
||||
now := at("2026-11-02T10:00:00Z")
|
||||
on := dbq.BackupSetting{Enabled: true}
|
||||
good := runTimes{lastSuccess: at("2026-11-02T03:00:00Z"), lastAttempt: at("2026-11-02T03:00:00Z")}
|
||||
cases := []struct {
|
||||
name string
|
||||
st dbq.BackupSetting
|
||||
probs []string
|
||||
t runTimes
|
||||
want string
|
||||
}{
|
||||
{"switched off, even with problems and failures", dbq.BackupSetting{}, []string{"x"}, runTimes{failingSince: now}, StateOff},
|
||||
{"on but not set up", on, []string{"x"}, good, StateIncomplete},
|
||||
{"the latest run failed", on, nil, runTimes{lastSuccess: good.lastSuccess, failingSince: now}, StateFailing},
|
||||
{"nothing has run yet", on, nil, runTimes{}, StateWaiting},
|
||||
{"the latest backup is old", on, nil, runTimes{lastSuccess: now.Add(-27 * time.Hour)}, StateOverdue},
|
||||
{"a day and a bit is still fine", on, nil, runTimes{lastSuccess: now.Add(-25 * time.Hour)}, StateOK},
|
||||
{"fine", on, nil, good, StateOK},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := verdict(c.st, c.probs, c.t, now); got != c.want {
|
||||
t.Errorf("%s: %s, want %s", c.name, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestProblemsNameWhatIsMissing(t *testing.T) {
|
||||
if p := problems(dbq.BackupSetting{Destination: DestinationS3}); len(p) != 4 {
|
||||
t.Errorf("an empty bucket: %v, want endpoint, name, access key and secret key", p)
|
||||
}
|
||||
if p := problems(dbq.BackupSetting{Destination: DestinationS3, Endpoint: "https://s3.example.com", Bucket: "b", AccessKey: "a", SecretKey: "v1.x"}); len(p) != 0 {
|
||||
t.Errorf("a whole bucket: %v", p)
|
||||
}
|
||||
if p := problems(dbq.BackupSetting{Destination: DestinationFolder}); len(p) != 1 {
|
||||
t.Errorf("an empty folder: %v", p)
|
||||
}
|
||||
if p := problems(dbq.BackupSetting{Destination: DestinationFolder, Folder: "/backups"}); len(p) != 0 {
|
||||
t.Errorf("a folder: %v", p)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeRefusesWhatCannotBeKept(t *testing.T) {
|
||||
ok := Update{Destination: DestinationS3, RunHour: 3}
|
||||
for field, u := range map[string]func(Update) Update{
|
||||
"destination": func(u Update) Update { u.Destination = "ftp"; return u },
|
||||
"run_hour": func(u Update) Update { u.RunHour = 24; return u },
|
||||
"upload_limit_kib": func(u Update) Update { u.UploadLimitKiB = -1; return u },
|
||||
"folder": func(u Update) Update { u.Folder = "backups"; return u },
|
||||
"endpoint": func(u Update) Update { u.Endpoint = "s3.example.com"; return u },
|
||||
} {
|
||||
_, err := merge(dbq.BackupSetting{}, u(ok))
|
||||
var fe *FieldError
|
||||
if !errors.As(err, &fe) || fe.Field != field {
|
||||
t.Errorf("%s: err = %v, want a FieldError on it", field, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeTidiesAndKeepsTheStoredSecret(t *testing.T) {
|
||||
st, err := merge(dbq.BackupSetting{SecretKey: "v1.stored"}, Update{
|
||||
Enabled: true, Destination: DestinationS3, Endpoint: " https://s3.example.com/ ",
|
||||
Bucket: " minstrel ", AccessKey: " ak ", RunHour: 4, UploadLimitKiB: 512, Folder: "/srv/backups/",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st.Endpoint != "https://s3.example.com" || st.Bucket != "minstrel" || st.AccessKey != "ak" ||
|
||||
st.SecretKey != "v1.stored" || st.RunHour != 4 || st.UploadLimitKib != 512 || st.Folder != "/srv/backups" || !st.Enabled {
|
||||
t.Errorf("merged = %+v", st)
|
||||
}
|
||||
}
|
||||
|
||||
func TestThrottleKeepsToItsRate(t *testing.T) {
|
||||
data := bytes.Repeat([]byte("x"), 20*1024)
|
||||
start := time.Now()
|
||||
got, err := io.ReadAll(Throttle(context.Background(), bytes.NewReader(data), 100*1024))
|
||||
if err != nil || !bytes.Equal(got, data) {
|
||||
t.Fatalf("read %d bytes, err %v", len(got), err)
|
||||
}
|
||||
// 20 KiB at 100 KiB a second takes at least a fifth of a second.
|
||||
if took := time.Since(start); took < 180*time.Millisecond {
|
||||
t.Errorf("took %s, want at least 200ms", took)
|
||||
}
|
||||
}
|
||||
|
||||
func TestThrottleWithNoLimitIsTheReaderItself(t *testing.T) {
|
||||
r := strings.NewReader("x")
|
||||
if Throttle(context.Background(), r, 0) != io.Reader(r) {
|
||||
t.Error("a zero limit wrapped the reader")
|
||||
}
|
||||
}
|
||||
|
||||
func TestThrottleStopsWhenItsContextEnds(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
_, err := io.ReadAll(Throttle(ctx, bytes.NewReader(make([]byte, 4096)), 1024))
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("err = %v, want context.Canceled", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDescribeSaysWhyARunStopped(t *testing.T) {
|
||||
if got := describe(context.DeadlineExceeded); !strings.Contains(got, "longer than 30 minutes") {
|
||||
t.Errorf("deadline: %q", got)
|
||||
}
|
||||
if got := describe(context.Canceled); !strings.Contains(got, "server stopped") {
|
||||
t.Errorf("canceled: %q", got)
|
||||
}
|
||||
if got := describe(errors.New("backup: the folder /x isn't there")); got != "the folder /x isn't there" {
|
||||
t.Errorf("other: %q", got)
|
||||
}
|
||||
}
|
||||
@@ -146,6 +146,9 @@ var Excluded = map[string]string{
|
||||
"want_candidates": "the last search's results; the next search finds them again",
|
||||
"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",
|
||||
}
|
||||
|
||||
// carried reports whether a table is in Tables.
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Throttle returns a reader that reads r no faster than perSecond bytes a
|
||||
// second, so an upload leaves room on the connection for everything else.
|
||||
// perSecond of 0 or less returns r unchanged. A wait ends early when ctx does,
|
||||
// and the read then returns ctx's error.
|
||||
func Throttle(ctx context.Context, r io.Reader, perSecond int64) io.Reader {
|
||||
if perSecond <= 0 {
|
||||
return r
|
||||
}
|
||||
return &throttled{ctx: ctx, r: r, perSecond: perSecond}
|
||||
}
|
||||
|
||||
type throttled struct {
|
||||
ctx context.Context
|
||||
r io.Reader
|
||||
perSecond int64
|
||||
start time.Time
|
||||
sent int64
|
||||
}
|
||||
|
||||
func (t *throttled) Read(p []byte) (int, error) {
|
||||
if t.start.IsZero() {
|
||||
t.start = time.Now()
|
||||
}
|
||||
// Reads are cut to a tenth of a second's worth, so the pace is even
|
||||
// rather than a burst followed by a long pause.
|
||||
if chunk := max(t.perSecond/10, 1); int64(len(p)) > chunk {
|
||||
p = p[:chunk]
|
||||
}
|
||||
n, err := t.r.Read(p)
|
||||
t.sent += int64(n)
|
||||
at := t.start.Add(time.Duration(float64(t.sent) / float64(t.perSecond) * float64(time.Second)))
|
||||
if wait := time.Until(at); wait > 0 {
|
||||
timer := time.NewTimer(wait)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-t.ctx.Done():
|
||||
return n, t.ctx.Err()
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.31.1
|
||||
// source: backups.sql
|
||||
|
||||
package dbq
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
const backupRunTimes = `-- name: BackupRunTimes :one
|
||||
SELECT
|
||||
COALESCE((SELECT max(started_at) FROM backup_runs WHERE status = 'succeeded'),
|
||||
'epoch'::timestamptz)::timestamptz AS last_success,
|
||||
COALESCE((SELECT max(started_at) FROM backup_runs),
|
||||
'epoch'::timestamptz)::timestamptz AS last_attempt,
|
||||
COALESCE((SELECT min(started_at) FROM backup_runs
|
||||
WHERE status = 'failed'
|
||||
AND started_at > COALESCE((SELECT max(started_at) FROM backup_runs WHERE status = 'succeeded'),
|
||||
'epoch'::timestamptz)),
|
||||
'epoch'::timestamptz)::timestamptz AS failing_since
|
||||
`
|
||||
|
||||
type BackupRunTimesRow struct {
|
||||
LastSuccess pgtype.Timestamptz
|
||||
LastAttempt pgtype.Timestamptz
|
||||
FailingSince pgtype.Timestamptz
|
||||
}
|
||||
|
||||
// The times the verdict turns on. A run that never happened is the epoch,
|
||||
// so the row always exists. failing_since is the first failure after the last
|
||||
// success, and the epoch when the latest finished run succeeded.
|
||||
func (q *Queries) BackupRunTimes(ctx context.Context) (BackupRunTimesRow, error) {
|
||||
row := q.db.QueryRow(ctx, backupRunTimes)
|
||||
var i BackupRunTimesRow
|
||||
err := row.Scan(&i.LastSuccess, &i.LastAttempt, &i.FailingSince)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const failUnfinishedBackupRuns = `-- name: FailUnfinishedBackupRuns :execrows
|
||||
UPDATE backup_runs
|
||||
SET status = 'failed', finished_at = now(),
|
||||
error = 'The server stopped while the backup was running.'
|
||||
WHERE status = 'running'
|
||||
`
|
||||
|
||||
func (q *Queries) FailUnfinishedBackupRuns(ctx context.Context) (int64, error) {
|
||||
result, err := q.db.Exec(ctx, failUnfinishedBackupRuns)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return result.RowsAffected(), nil
|
||||
}
|
||||
|
||||
const finishBackupRun = `-- name: FinishBackupRun :exec
|
||||
UPDATE backup_runs
|
||||
SET status = $2,
|
||||
finished_at = now(),
|
||||
location = $3,
|
||||
bytes = $4,
|
||||
error = $5
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type FinishBackupRunParams struct {
|
||||
ID pgtype.UUID
|
||||
Status string
|
||||
Location string
|
||||
Bytes int64
|
||||
Error string
|
||||
}
|
||||
|
||||
func (q *Queries) FinishBackupRun(ctx context.Context, arg FinishBackupRunParams) error {
|
||||
_, err := q.db.Exec(ctx, finishBackupRun,
|
||||
arg.ID,
|
||||
arg.Status,
|
||||
arg.Location,
|
||||
arg.Bytes,
|
||||
arg.Error,
|
||||
)
|
||||
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
|
||||
`
|
||||
|
||||
// Minstrel's own backup: its settings and its runs.
|
||||
func (q *Queries) GetBackupSettings(ctx context.Context) (BackupSetting, error) {
|
||||
row := q.db.QueryRow(ctx, getBackupSettings)
|
||||
var i BackupSetting
|
||||
err := row.Scan(
|
||||
&i.ID,
|
||||
&i.Enabled,
|
||||
&i.Destination,
|
||||
&i.Folder,
|
||||
&i.Endpoint,
|
||||
&i.Bucket,
|
||||
&i.AccessKey,
|
||||
&i.SecretKey,
|
||||
&i.RunHour,
|
||||
&i.UploadLimitKib,
|
||||
&i.UpdatedAt,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const listBackupRuns = `-- name: ListBackupRuns :many
|
||||
SELECT id, trigger, status, started_at, finished_at, location, bytes, error
|
||||
FROM backup_runs
|
||||
ORDER BY started_at DESC
|
||||
LIMIT $1
|
||||
`
|
||||
|
||||
func (q *Queries) ListBackupRuns(ctx context.Context, limit int32) ([]BackupRun, error) {
|
||||
rows, err := q.db.Query(ctx, listBackupRuns, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []BackupRun
|
||||
for rows.Next() {
|
||||
var i BackupRun
|
||||
if err := rows.Scan(
|
||||
&i.ID,
|
||||
&i.Trigger,
|
||||
&i.Status,
|
||||
&i.StartedAt,
|
||||
&i.FinishedAt,
|
||||
&i.Location,
|
||||
&i.Bytes,
|
||||
&i.Error,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const pruneBackupRuns = `-- name: PruneBackupRuns :execrows
|
||||
DELETE FROM backup_runs WHERE status <> 'running' AND started_at < $1
|
||||
`
|
||||
|
||||
func (q *Queries) PruneBackupRuns(ctx context.Context, startedAt pgtype.Timestamptz) (int64, error) {
|
||||
result, err := q.db.Exec(ctx, pruneBackupRuns, startedAt)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return result.RowsAffected(), nil
|
||||
}
|
||||
|
||||
const saveBackupSettings = `-- name: SaveBackupSettings :exec
|
||||
UPDATE backup_settings
|
||||
SET enabled = $1,
|
||||
destination = $2,
|
||||
folder = $3,
|
||||
endpoint = $4,
|
||||
bucket = $5,
|
||||
access_key = $6,
|
||||
secret_key = $7,
|
||||
run_hour = $8,
|
||||
upload_limit_kib = $9,
|
||||
updated_at = now()
|
||||
WHERE id = true
|
||||
`
|
||||
|
||||
type SaveBackupSettingsParams struct {
|
||||
Enabled bool
|
||||
Destination string
|
||||
Folder string
|
||||
Endpoint string
|
||||
Bucket string
|
||||
AccessKey string
|
||||
SecretKey string
|
||||
RunHour int16
|
||||
UploadLimitKib int32
|
||||
}
|
||||
|
||||
func (q *Queries) SaveBackupSettings(ctx context.Context, arg SaveBackupSettingsParams) error {
|
||||
_, err := q.db.Exec(ctx, saveBackupSettings,
|
||||
arg.Enabled,
|
||||
arg.Destination,
|
||||
arg.Folder,
|
||||
arg.Endpoint,
|
||||
arg.Bucket,
|
||||
arg.AccessKey,
|
||||
arg.SecretKey,
|
||||
arg.RunHour,
|
||||
arg.UploadLimitKib,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const startBackupRun = `-- name: StartBackupRun :one
|
||||
INSERT INTO backup_runs (trigger) VALUES ($1)
|
||||
RETURNING id, started_at
|
||||
`
|
||||
|
||||
type StartBackupRunRow struct {
|
||||
ID pgtype.UUID
|
||||
StartedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
func (q *Queries) StartBackupRun(ctx context.Context, trigger string) (StartBackupRunRow, error) {
|
||||
row := q.db.QueryRow(ctx, startBackupRun, trigger)
|
||||
var i StartBackupRunRow
|
||||
err := row.Scan(&i.ID, &i.StartedAt)
|
||||
return i, err
|
||||
}
|
||||
@@ -305,6 +305,31 @@ type AuditLog struct {
|
||||
CreatedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
type BackupRun struct {
|
||||
ID pgtype.UUID
|
||||
Trigger string
|
||||
Status string
|
||||
StartedAt pgtype.Timestamptz
|
||||
FinishedAt pgtype.Timestamptz
|
||||
Location string
|
||||
Bytes int64
|
||||
Error string
|
||||
}
|
||||
|
||||
type BackupSetting struct {
|
||||
ID bool
|
||||
Enabled bool
|
||||
Destination string
|
||||
Folder string
|
||||
Endpoint string
|
||||
Bucket string
|
||||
AccessKey string
|
||||
SecretKey string
|
||||
RunHour int16
|
||||
UploadLimitKib int32
|
||||
UpdatedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
type CandidateArtistTag struct {
|
||||
CandidateMbid string
|
||||
Tag string
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE IF EXISTS backup_runs;
|
||||
DROP TABLE IF EXISTS backup_settings;
|
||||
@@ -0,0 +1,44 @@
|
||||
-- 0099_backups.up.sql — Minstrel backing itself up.
|
||||
--
|
||||
-- backup_settings one row: whether the nightly backup is on, where it goes
|
||||
-- (a folder, or an S3 bucket such as a Fabled Archives
|
||||
-- target), and the hour it runs, in UTC. Off until an
|
||||
-- admin switches it on.
|
||||
-- backup_runs one row per attempt: when, what started it, how it ended,
|
||||
-- where the file went, its size, and the error in words.
|
||||
CREATE TABLE backup_settings (
|
||||
id boolean PRIMARY KEY DEFAULT true CONSTRAINT backup_settings_one_row CHECK (id),
|
||||
enabled boolean NOT NULL DEFAULT false,
|
||||
destination text NOT NULL DEFAULT 's3'
|
||||
CONSTRAINT backup_settings_destination_check CHECK (destination IN ('folder', 's3')),
|
||||
folder text NOT NULL DEFAULT '',
|
||||
endpoint text NOT NULL DEFAULT '',
|
||||
bucket text NOT NULL DEFAULT '',
|
||||
access_key text NOT NULL DEFAULT '',
|
||||
-- Sealed under the stream secret, as every stored credential is.
|
||||
secret_key text NOT NULL DEFAULT '',
|
||||
run_hour smallint NOT NULL DEFAULT 3
|
||||
CONSTRAINT backup_settings_run_hour_check CHECK (run_hour BETWEEN 0 AND 23),
|
||||
-- The most an upload to the bucket may send, in KiB a second; 0 is no limit.
|
||||
upload_limit_kib integer NOT NULL DEFAULT 4096
|
||||
CONSTRAINT backup_settings_upload_limit_check CHECK (upload_limit_kib >= 0),
|
||||
updated_at timestamptz NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
INSERT INTO backup_settings DEFAULT VALUES;
|
||||
|
||||
CREATE TABLE backup_runs (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
trigger text NOT NULL
|
||||
CONSTRAINT backup_runs_trigger_check CHECK (trigger IN ('schedule', 'manual')),
|
||||
status text NOT NULL DEFAULT 'running'
|
||||
CONSTRAINT backup_runs_status_check CHECK (status IN ('running', 'succeeded', 'failed')),
|
||||
started_at timestamptz NOT NULL DEFAULT now(),
|
||||
finished_at timestamptz,
|
||||
-- Where the file went: the object key, or the file's path in the folder.
|
||||
location text NOT NULL DEFAULT '',
|
||||
bytes bigint NOT NULL DEFAULT 0,
|
||||
error text NOT NULL DEFAULT ''
|
||||
);
|
||||
|
||||
CREATE INDEX backup_runs_started_at_idx ON backup_runs (started_at DESC);
|
||||
@@ -0,0 +1,61 @@
|
||||
-- Minstrel's own backup: its settings and its runs.
|
||||
|
||||
-- name: GetBackupSettings :one
|
||||
SELECT * FROM backup_settings WHERE id = true;
|
||||
|
||||
-- name: SaveBackupSettings :exec
|
||||
UPDATE backup_settings
|
||||
SET enabled = $1,
|
||||
destination = $2,
|
||||
folder = $3,
|
||||
endpoint = $4,
|
||||
bucket = $5,
|
||||
access_key = $6,
|
||||
secret_key = $7,
|
||||
run_hour = $8,
|
||||
upload_limit_kib = $9,
|
||||
updated_at = now()
|
||||
WHERE id = true;
|
||||
|
||||
-- name: StartBackupRun :one
|
||||
INSERT INTO backup_runs (trigger) VALUES ($1)
|
||||
RETURNING id, started_at;
|
||||
|
||||
-- name: FinishBackupRun :exec
|
||||
UPDATE backup_runs
|
||||
SET status = $2,
|
||||
finished_at = now(),
|
||||
location = $3,
|
||||
bytes = $4,
|
||||
error = $5
|
||||
WHERE id = $1;
|
||||
|
||||
-- name: ListBackupRuns :many
|
||||
SELECT id, trigger, status, started_at, finished_at, location, bytes, error
|
||||
FROM backup_runs
|
||||
ORDER BY started_at DESC
|
||||
LIMIT $1;
|
||||
|
||||
-- name: BackupRunTimes :one
|
||||
-- The times the verdict turns on. A run that never happened is the epoch,
|
||||
-- so the row always exists. failing_since is the first failure after the last
|
||||
-- success, and the epoch when the latest finished run succeeded.
|
||||
SELECT
|
||||
COALESCE((SELECT max(started_at) FROM backup_runs WHERE status = 'succeeded'),
|
||||
'epoch'::timestamptz)::timestamptz AS last_success,
|
||||
COALESCE((SELECT max(started_at) FROM backup_runs),
|
||||
'epoch'::timestamptz)::timestamptz AS last_attempt,
|
||||
COALESCE((SELECT min(started_at) FROM backup_runs
|
||||
WHERE status = 'failed'
|
||||
AND started_at > COALESCE((SELECT max(started_at) FROM backup_runs WHERE status = 'succeeded'),
|
||||
'epoch'::timestamptz)),
|
||||
'epoch'::timestamptz)::timestamptz AS failing_since;
|
||||
|
||||
-- name: FailUnfinishedBackupRuns :execrows
|
||||
UPDATE backup_runs
|
||||
SET status = 'failed', finished_at = now(),
|
||||
error = 'The server stopped while the backup was running.'
|
||||
WHERE status = 'running';
|
||||
|
||||
-- name: PruneBackupRuns :execrows
|
||||
DELETE FROM backup_runs WHERE status <> 'running' AND started_at < $1;
|
||||
@@ -184,6 +184,16 @@ func ResetDB(t *testing.T, pool *pgxpool.Pool) {
|
||||
); err != nil {
|
||||
t.Fatalf("dbtest.ResetDB reset release-type profiles: %v", err)
|
||||
}
|
||||
// The server's own backup: no runs, and settings back to off.
|
||||
if _, err := pool.Exec(ctx, `
|
||||
DELETE FROM backup_runs;
|
||||
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`,
|
||||
); err != nil {
|
||||
t.Fatalf("dbtest.ResetDB reset backup settings: %v", err)
|
||||
}
|
||||
// Search and grab settings, a singleton reset the same way.
|
||||
// The library listing reads as done, so a test's albums are not
|
||||
// put on the wanted list behind its back; its own tests clear it.
|
||||
|
||||
@@ -35,6 +35,7 @@ var (
|
||||
SubsonicPassword = Column{"users", "subsonic_password", "id", false}
|
||||
CoverArtAPIKey = Column{"cover_art_provider_settings", "api_key", "provider_id", false}
|
||||
TagProviderAPIKey = Column{"tag_provider_settings", "api_key", "provider_id", false}
|
||||
BackupSecretKey = Column{"backup_settings", "secret_key", "id", true}
|
||||
)
|
||||
|
||||
// Columns is every sealed column, for SealStored and for anything that has to
|
||||
@@ -42,6 +43,7 @@ var (
|
||||
var Columns = []Column{
|
||||
AcquisitionAPIKey, AcquisitionPassword, AcoustIDAPIKey, SMTPPassword,
|
||||
ListenBrainzToken, SubsonicPassword, CoverArtAPIKey, TagProviderAPIKey,
|
||||
BackupSecretKey,
|
||||
}
|
||||
|
||||
// Owner is the owner string for a row whose key is key. A single-row table's
|
||||
|
||||
@@ -84,6 +84,7 @@ func TestSealStoredSealsWhatIsInTheClear_Integration(t *testing.T) {
|
||||
`INSERT INTO tag_provider_settings (provider_id, enabled, display_order, api_key)
|
||||
VALUES ('seal-test', false, 99, 'tag-key')
|
||||
ON CONFLICT (provider_id) DO UPDATE SET api_key = EXCLUDED.api_key`,
|
||||
`UPDATE backup_settings SET secret_key = 'bucket-secret' WHERE id = true`,
|
||||
}
|
||||
for _, s := range seed {
|
||||
if _, err := pool.Exec(ctx, s); err != nil {
|
||||
@@ -97,6 +98,7 @@ func TestSealStoredSealsWhatIsInTheClear_Integration(t *testing.T) {
|
||||
`DELETE FROM acquisition_services WHERE code = 'seal-test'`,
|
||||
`UPDATE acoustid_settings SET api_key = NULL WHERE id = true`,
|
||||
`UPDATE smtp_config SET password = '' WHERE id = true`,
|
||||
`UPDATE backup_settings SET secret_key = '' WHERE id = true`,
|
||||
} {
|
||||
_, _ = pool.Exec(context.Background(), s)
|
||||
}
|
||||
@@ -115,6 +117,7 @@ func TestSealStoredSealsWhatIsInTheClear_Integration(t *testing.T) {
|
||||
{secret.SubsonicPassword, u.ID.String(), "sub-pw"},
|
||||
{secret.CoverArtAPIKey, "seal-test", "cover-key"},
|
||||
{secret.TagProviderAPIKey, "seal-test", "tag-key"},
|
||||
{secret.BackupSecretKey, "true", "bucket-secret"},
|
||||
}
|
||||
if len(want) != len(secret.Columns) {
|
||||
t.Fatalf("the test covers %d columns, secret.Columns lists %d", len(want), len(secret.Columns))
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/api"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/auth"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/backup"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/buildinfo"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/config"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/coverart"
|
||||
@@ -94,6 +95,9 @@ type Server struct {
|
||||
// the admin card reaches it without a restart. Router() constructs a
|
||||
// fallback when nil (tests).
|
||||
ReacqSettings *reacquisition.SettingsService
|
||||
// Backups is the server's own backup, the instance whose loop runs the
|
||||
// nightly schedule. Nil builds one that runs only when an admin asks.
|
||||
Backups *backup.Service
|
||||
// StreamSecret is the HMAC key used by /api/cast/stream-token to
|
||||
// mint signed UPnP / Sonos stream URLs and by /api/tracks/{id}/stream
|
||||
// to verify them. Sourced from config.Config.StreamSecret. Tests that
|
||||
@@ -200,7 +204,11 @@ func (s *Server) Router() http.Handler {
|
||||
}
|
||||
acoustIDLookup = library.NewAcoustIDLookupWorker(s.Pool, s.Logger, acoustIDSettings, nil)
|
||||
}
|
||||
api.Mount(r, s.Pool, s.Logger, writer, s.RecommendationCfg, recSettings, requestSvc, quarantineSvc, tracksSvc, playlistsSvc, s.CoverEnricher, s.CoverSettings, s.TagSettings, s.LibraryScanner, s.ScanCfg, s.DataDir, smtpSender, bus, s.PlaylistScheduler, s.StreamSecret, netSettings, reacqSettings, fpSettings, loudSettings, acoustIDLookup)
|
||||
backups := s.Backups
|
||||
if backups == nil {
|
||||
backups = backup.NewService(s.Pool, s.Logger, buildinfo.Version)
|
||||
}
|
||||
api.Mount(r, s.Pool, s.Logger, writer, s.RecommendationCfg, recSettings, requestSvc, quarantineSvc, tracksSvc, playlistsSvc, s.CoverEnricher, s.CoverSettings, s.TagSettings, s.LibraryScanner, s.ScanCfg, s.DataDir, smtpSender, bus, s.PlaylistScheduler, s.StreamSecret, netSettings, reacqSettings, fpSettings, loudSettings, acoustIDLookup, backups)
|
||||
// /api/admin/scan is the only admin route owned by the server package
|
||||
// (it needs the Scanner). Register it as a single inline-middleware
|
||||
// route — using r.Route("/api/admin", ...) here would create a second
|
||||
|
||||
Reference in New Issue
Block a user