michael/URBM
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add pause resume and abort controls

Browse Source
This commit is contained in:
Mikei386
2026-06-14 22:34:22 +02:00
parent 82d00bd116
commit 656d569b27
13 changed files with 228 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/server.go
+18
View File
@@ -41,6 +41,8 @@ func New(socket string, svc *service.Service, log *slog.Logger) *Server {
mux.HandleFunc("GET /v1/workloads/{kind}", s.workloads)
mux.HandleFunc("POST /v1/jobs/{id}/run", s.runJob)
mux.HandleFunc("POST /v1/runs/{id}/cancel", s.cancelRun)
mux.HandleFunc("POST /v1/runs/{id}/pause", s.pauseRun)
mux.HandleFunc("POST /v1/runs/{id}/resume", s.resumeRun)
mux.HandleFunc("POST /v1/repositories/{id}/test", s.testRepository)
mux.HandleFunc("POST /v1/repositories/{id}/init", s.initRepository)
mux.HandleFunc("POST /v1/repositories/{id}/{action}", s.maintenance)
@@ -157,6 +159,22 @@ func (s *Server) cancelRun(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(204)
}
func (s *Server) pauseRun(w http.ResponseWriter, r *http.Request) {
if err := s.service.Pause(r.PathValue("id")); err != nil {
writeError(w, err)
return
}
w.WriteHeader(204)
}
func (s *Server) resumeRun(w http.ResponseWriter, r *http.Request) {
if err := s.service.Resume(r.PathValue("id")); err != nil {
writeError(w, err)
return
}
w.WriteHeader(204)
}
func (s *Server) testRepository(w http.ResponseWriter, r *http.Request) {
s.repositoryAction(w, r, false)
}
internal/queue/queue.go
+20 -1
View File
@@ -37,7 +37,7 @@ func (q *Queue) RestoreHistory(runs []model.Run) {
q.mu.Lock()
defer q.mu.Unlock()
for _, run := range runs {
if run.Status == "queued" || run.Status == "running" {
if run.Status == "queued" || run.Status == "running" || run.Status == "paused" {
run.Status = "failed"
run.ErrorCode = "environment"
run.Message = "daemon restarted while task was active"
@@ -125,6 +125,25 @@ func (q *Queue) Cancel(id string) bool {
return false
}
func (q *Queue) SetPaused(id string, paused bool) bool {
q.mu.Lock()
defer q.mu.Unlock()
if q.active == nil || q.active.ID != id {
return false
}
expected := "running"
next := "paused"
if !paused {
expected, next = "paused", "running"
}
if q.active.Status != expected {
return false
}
q.active.Status = next
q.emitLocked()
return true
}
func (q *Queue) Stop() {
q.mu.Lock()
q.stopping = true
internal/queue/queue_test.go
+31
View File
@@ -59,3 +59,34 @@ func TestClearHistoryKeepsActiveAndPendingRuns(t *testing.T) {
q.Stop()
close(release)
}
func TestPauseAndResumeActiveRun(t *testing.T) {
started := make(chan struct{})
release := make(chan struct{})
q := New(func(_ context.Context, run model.Run) model.Run {
close(started)
<-release
run.Status = "success"
return run
}, nil)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go q.Run(ctx)
if err := q.Enqueue(model.Run{ID: "run", JobID: "job"}); err != nil {
t.Fatal(err)
}
<-started
if !q.SetPaused("run", true) {
t.Fatal("active run could not be paused")
}
if runs := q.Snapshot(); len(runs) != 1 || runs[0].Status != "paused" {
t.Fatalf("paused runs = %#v", runs)
}
if !q.SetPaused("run", false) {
t.Fatal("paused run could not be resumed")
}
if runs := q.Snapshot(); len(runs) != 1 || runs[0].Status != "running" {
t.Fatalf("resumed runs = %#v", runs)
}
close(release)
}
internal/restic/restic.go
+59
View File
@@ -12,6 +12,8 @@ import (
"os/exec"
"path/filepath"
"strings"
"sync"
"syscall"
"github.com/backupper-unraid/backupper/internal/model"
)
@@ -23,6 +25,47 @@ type Runner struct {
RuntimeDir string
Secrets SecretGetter
Log *slog.Logger
mu sync.Mutex
processes map[string]*os.Process
}
type runIDKey struct{}
func WithRunID(ctx context.Context, runID string) context.Context {
return context.WithValue(ctx, runIDKey{}, runID)
}
func (r *Runner) Pause(runID string) error { return r.signal(runID, syscall.SIGSTOP) }
func (r *Runner) Resume(runID string) error { return r.signal(runID, syscall.SIGCONT) }
func (r *Runner) signal(runID string, signal syscall.Signal) error {
r.mu.Lock()
process := r.processes[runID]
r.mu.Unlock()
if process == nil {
return errors.New("restic process is not active yet")
}
if err := syscall.Kill(-process.Pid, signal); err != nil {
return fmt.Errorf("signal restic process group: %w", err)
}
return nil
}
func (r *Runner) registerProcess(runID string, process *os.Process) func() {
if runID == "" {
return func() {}
}
r.mu.Lock()
if r.processes == nil {
r.processes = make(map[string]*os.Process)
}
r.processes[runID] = process
r.mu.Unlock()
return func() {
r.mu.Lock()
delete(r.processes, runID)
r.mu.Unlock()
}
}
type Summary struct {
@@ -203,6 +246,19 @@ func (r *Runner) run(ctx context.Context, repo model.Repository, args []string,
}
fullArgs := append(globalArgs, args...)
cmd := exec.CommandContext(ctx, r.Binary, fullArgs...)
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
cmd.Cancel = func() error {
if cmd.Process == nil {
return os.ErrProcessDone
}
if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL); err != nil {
if errors.Is(err, syscall.ESRCH) {
return os.ErrProcessDone
}
return err
}
return nil
}
cmd.Env = []string{"PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "HOME=/root", "RESTIC_PASSWORD_FILE=" + passwordPath}
stdout, err := cmd.StdoutPipe()
if err != nil {
@@ -215,6 +271,9 @@ func (r *Runner) run(ctx context.Context, repo model.Repository, args []string,
if err := cmd.Start(); err != nil {
return classify(err)
}
runID, _ := ctx.Value(runIDKey{}).(string)
unregister := r.registerProcess(runID, cmd.Process)
defer unregister()
var captured []byte
done := make(chan struct{})
go func() {
internal/restic/restic_test.go
+58
View File
@@ -7,6 +7,7 @@ import (
"path/filepath"
"strings"
"testing"
"time"
"github.com/backupper-unraid/backupper/internal/model"
)
@@ -88,3 +89,60 @@ func TestResticErrorExplainsCiphertextVerification(t *testing.T) {
t.Fatalf("unexpected error: %v", err)
}
}
func TestRunnerPausesResumesAndCancelsProcessGroup(t *testing.T) {
dir := t.TempDir()
ticksPath := filepath.Join(dir, "ticks")
script := filepath.Join(dir, "restic")
body := fmt.Sprintf("#!/bin/sh\nwhile :; do printf x >> '%s'; sleep 0.05; done\n", ticksPath)
if err := os.WriteFile(script, []byte(body), 0700); err != nil {
t.Fatal(err)
}
runner := &Runner{Binary: script, RuntimeDir: dir, Secrets: fakeSecrets{"password": "secret"}}
repo := model.Repository{Type: model.RepositoryLocal, Location: "/repo", PasswordRef: "password"}
ctx, cancel := context.WithCancel(WithRunID(context.Background(), "run"))
done := make(chan error, 1)
go func() { done <- runner.run(ctx, repo, []string{"backup"}, nil, nil) }()
waitForSize := func(minimum int64) int64 {
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if info, err := os.Stat(ticksPath); err == nil && info.Size() >= minimum {
return info.Size()
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("tick file did not reach %d bytes", minimum)
return 0
}
waitForSize(2)
if err := runner.Pause("run"); err != nil {
t.Fatal(err)
}
time.Sleep(100 * time.Millisecond)
pausedSize := waitForSize(1)
time.Sleep(150 * time.Millisecond)
info, err := os.Stat(ticksPath)
if err != nil {
t.Fatal(err)
}
if info.Size() != pausedSize {
t.Fatalf("process continued while paused: size=%d, expected=%d", info.Size(), pausedSize)
}
if err := runner.Resume("run"); err != nil {
t.Fatal(err)
}
waitForSize(pausedSize + 2)
if err := runner.Pause("run"); err != nil {
t.Fatal(err)
}
cancel()
select {
case err := <-done:
if err == nil || !strings.Contains(err.Error(), "cancelled") {
t.Fatalf("cancel result = %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("paused process group was not cancelled")
}
}
internal/service/service.go
+22
View File
@@ -168,6 +168,27 @@ func (s *Service) EnqueueRestore(task model.RestoreTask) (model.Run, error) {
func (s *Service) Cancel(runID string) bool { return s.queue.Cancel(runID) }
func (s *Service) ClearRunHistory() int { return s.queue.ClearHistory() }
func (s *Service) Pause(runID string) error {
if err := s.restic.Pause(runID); err != nil {
return fmt.Errorf("pause: %w", err)
}
if !s.queue.SetPaused(runID, true) {
_ = s.restic.Resume(runID)
return errors.New("pause: run is not currently running")
}
return nil
}
func (s *Service) Resume(runID string) error {
if err := s.restic.Resume(runID); err != nil {
return fmt.Errorf("resume: %w", err)
}
if !s.queue.SetPaused(runID, false) {
return errors.New("resume: run is not currently paused")
}
return nil
}
func (s *Service) Snapshots(ctx context.Context, repoID string) ([]model.Snapshot, error) {
repo, ok := s.repository(repoID)
if !ok {
@@ -212,6 +233,7 @@ func (s *Service) TestRepository(ctx context.Context, repoID string, initialize
}
func (s *Service) execute(ctx context.Context, run model.Run) model.Run {
ctx = restic.WithRunID(ctx, run.ID)
if run.TaskType == "backup" {
return s.executeBackup(ctx, run)
}