proyectos/move-data-nas
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase A-E: stability, security, observability, and test coverage

Browse Source 
Phase A - Stability:
- Engine.Start(): recovers orphaned jobs (running/queued/waking_up) after crash
- Engine.Stop(): graceful shutdown - cancels all in-flight jobs and waits
- Queue keyed by jobID (not syncPairID): cancel now targets exact job
- Local rsync uses jobCtx (context.Background() replaced)
- Migrations wrapped in transactions; checksums stored

Phase B - Security:
- admin/admin default removed: SYNCSERVER_ADMIN_PASSWORD required on first run
- Path validation: rejects .., leading -, null bytes in sync pair paths
- Rsync flags allowlist: dangerous flags blocked (--rsync-path, -e, --files-from)
- Shell concat in RunRemote replaced with proper sh -c escaping
- knownhosts: replaced custom parser with golang.org/x/crypto/ssh/knownhosts
- RequireAdmin wired: machine CRUD, SSH key ops, settings require admin role
- deploy-keys: uses authorized_keys only (no private key upload)
- Hardcoded /var/lib/syncserver/ssh paths replaced with cfg.SSHDir()

Phase C - Operational:
- /readyz health check: DB query + SSH dir accessibility
- /metrics endpoint: Prometheus text format (jobs, queue, machines)
- Event struct JSON tags: job_id, machine_id, type (snake_case)
- EventBus broadcast: fanned out to all subscribers
- SQLite VACUUM INTO backup: scheduled before cleanup if BackupDir set
- Filesystem job log cleanup: removes .log files for purged jobs
- Backup retention: old backups auto-purged

Phase D - Frontend:
- Schedules page: REST API + full CRUD UI for cron schedules
- Dashboard: cancel button for running/queued jobs
- JobDetail: server-side log download via API
- Settings: displays data_dir from server
- 404 page: proper NotFound component

Phase E - Tests:
- auth_test.go: JWT, bcrypt, middleware, seed (18 tests)
- models_test.go: Job, Machine, SyncPair, Schedule repos (18 tests)
- go test -race: no data races found
This commit is contained in:
Daniel Arroyo
2026-07-19 22:14:30 -04:00
parent 300555d35f
commit 84b185be39
33 changed files with 2398 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/syncengine/engine.go
+118 -29
View File
@@ -18,26 +18,31 @@ import (
)
type Engine struct {
db *sql.DB
cfg *config.Config
queue *Queue
eventBus *EventBus
mu sync.RWMutex
stopped bool
lastProbeAt atomic.Int64
db *sql.DB
cfg *config.Config
queue *Queue
eventBus *EventBus
mu sync.RWMutex
stopped bool
lastProbeAt atomic.Int64
jobsWG sync.WaitGroup
stopCh chan struct{}
jobsTotal map[string]int64
jobsTotalMu sync.Mutex
}
type Event struct {
Type string
JobID int64
MachineID int64
Key string
Value string
Line string
Stream string
Progress *ProgressFields
TotalBytes int64
SentBytes int64
Type string `json:"type"`
JobID int64 `json:"job_id"`
MachineID int64 `json:"machine_id"`
Key string `json:"key,omitempty"`
Value string `json:"value,omitempty"`
Line string `json:"line,omitempty"`
Stream string `json:"stream,omitempty"`
Progress *ProgressFields `json:"progress,omitempty"`
TotalBytes int64 `json:"total_bytes,omitempty"`
SentBytes int64 `json:"sent_bytes,omitempty"`
}
func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine {
@@ -46,12 +51,56 @@ func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine {
cfg: cfg,
queue: NewQueue(),
eventBus: NewEventBus(200),
stopCh: make(chan struct{}),
jobsTotal: map[string]int64{
"success": 0,
"failed": 0,
"cancelled": 0,
},
}
return e
}
func (e *Engine) Start() {}
func (e *Engine) Stop() {}
func (e *Engine) Start() {
e.recoverOrphanedJobs()
slog.Info("engine started")
}
func (e *Engine) Stop() {
e.mu.Lock()
if e.stopped {
e.mu.Unlock()
return
}
e.stopped = true
e.mu.Unlock()
close(e.stopCh)
runningIDs := e.queue.RunningJobs()
for _, id := range runningIDs {
e.queue.Cancel(id, false)
}
e.jobsWG.Wait()
slog.Info("engine stopped")
}
func (e *Engine) recoverOrphanedJobs() {
jobRepo := models.NewJobRepository(e.db)
jobs, err := jobRepo.GetByStatusAny([]string{"queued", "waking_up", "running"})
if err != nil {
slog.Warn("failed to recover orphaned jobs", "error", err)
return
}
for _, j := range jobs {
slog.Warn("recovered orphaned job, marking as failed",
"job_id", j.ID, "pair_id", j.SyncPairID, "status", j.Status)
jobRepo.UpdateStatus(j.ID, "failed")
jobRepo.SetError(j.ID, "crash_recovery",
fmt.Sprintf("job was %s when server shut down unexpectedly", j.Status))
}
}
func (e *Engine) SubscribeJob(jobID int64) (chan Event, func()) {
return e.eventBus.Subscribe(jobID)
@@ -84,9 +133,12 @@ func (e *Engine) wakeMachine(ctx context.Context, m *models.Machine) {
}
func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
if e.queue.IsRunning(pairID) {
existingJobID, _ := e.queue.GetJobID(pairID)
return fmt.Errorf("%w: job %d is already running", ErrAlreadyRunning, existingJobID)
if e.queue.IsRunning(jobID) {
return fmt.Errorf("%w: job %d is already running", ErrAlreadyRunning, jobID)
}
if existingJobID, exists := e.queue.GetByPair(pairID); exists {
return fmt.Errorf("%w: job %d is already running for this sync pair", ErrAlreadyRunning, existingJobID)
}
jobCtx, cancel := context.WithCancel(ctx)
@@ -99,7 +151,10 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
if enqueueErr != nil {
return enqueueErr
}
defer e.queue.Dequeue(pairID)
defer e.queue.Dequeue(jobID)
e.jobsWG.Add(1)
defer e.jobsWG.Done()
pairRepo := models.NewSyncPairRepository(e.db)
pair, err := pairRepo.GetByID(pairID)
@@ -381,17 +436,18 @@ func (e *Engine) resolveSSHKey(machine *models.Machine) (string, error) {
return sshKey.PrivateKeyPath, nil
}
func (e *Engine) Cancel(jobID int64, syncPairID int64, byUser bool) bool {
if e.queue.IsRunning(syncPairID) {
e.queue.Cancel(syncPairID, byUser)
return true
}
return false
func (e *Engine) Cancel(jobID int64, byUser bool) bool {
return e.queue.Cancel(jobID, byUser)
}
func (e *Engine) setJobStatus(jobID int64, status string) {
jobRepo := models.NewJobRepository(e.db)
jobRepo.UpdateStatus(jobID, status)
if status == "success" || status == "failed" || status == "cancelled" {
e.jobsTotalMu.Lock()
e.jobsTotal[status]++
e.jobsTotalMu.Unlock()
}
}
func (e *Engine) setJobLogFile(jobID int64, path string) {
@@ -495,3 +551,36 @@ func (e *Engine) ProbeAllMachines() {
}
wg.Wait()
}
func (e *Engine) GetJobsTotal() map[string]int64 {
e.jobsTotalMu.Lock()
defer e.jobsTotalMu.Unlock()
return e.jobsTotal
}
func (e *Engine) GetJobsRunning() int64 {
return int64(len(e.queue.RunningJobs()))
}
func (e *Engine) GetQueueDepth() int64 {
return int64(e.queue.Len())
}
func (e *Engine) GetMachineCounts() (online, total int64) {
machineRepo := models.NewMachineRepository(e.db)
ms, err := machineRepo.GetAll()
if err != nil {
return 0, 0
}
for _, m := range ms {
total++
if m.Status == "online" {
online++
}
}
return online, total
}
func (e *Engine) DB() *sql.DB {
return e.db
}
internal/syncengine/eventbus.go
+9 -31
View File
@@ -8,20 +8,17 @@ import (
type EventBus struct {
subscribers map[int64]map[chan Event]struct{}
mu sync.RWMutex
global chan Event
bufferSize int
globalSubs []globalSub
}
type globalSub struct {
ch chan Event
done chan struct{}
ch chan Event
}
func NewEventBus(bufferSize int) *EventBus {
return &EventBus{
subscribers: make(map[int64]map[chan Event]struct{}),
global: make(chan Event, bufferSize),
bufferSize: bufferSize,
globalSubs: nil,
}
@@ -51,32 +48,10 @@ func (eb *EventBus) Subscribe(jobID int64) (chan Event, func()) {
func (eb *EventBus) SubscribeGlobal() (chan Event, func()) {
ch := make(chan Event, eb.bufferSize)
done := make(chan struct{})
eb.mu.Lock()
eb.globalSubs = append(eb.globalSubs, globalSub{ch: ch, done: done})
eb.globalSubs = append(eb.globalSubs, globalSub{ch: ch})
eb.mu.Unlock()
go func() {
defer func() {
if r := recover(); r != nil {
slog.Error("SubscribeGlobal goroutine panicked", "reason", r)
}
close(ch)
}()
for {
select {
case evt := <-eb.global:
select {
case ch <- evt:
default:
slog.Warn("global event subscriber buffer full, dropping event", "type", evt.Type)
}
case <-done:
return
}
}
}()
return ch, func() {
close(done)
eb.mu.Lock()
for i, s := range eb.globalSubs {
if s.ch == ch {
@@ -85,6 +60,7 @@ func (eb *EventBus) SubscribeGlobal() (chan Event, func()) {
}
}
eb.mu.Unlock()
close(ch)
}
}
@@ -102,10 +78,12 @@ func (eb *EventBus) Publish(evt Event) {
}
}
select {
case eb.global <- evt:
default:
slog.Warn("global event bus full, dropping event", "type", evt.Type)
for _, sub := range eb.globalSubs {
select {
case sub.ch <- evt:
default:
slog.Warn("global event subscriber buffer full, dropping event", "type", evt.Type)
}
}
}
internal/syncengine/parser.go
+14 -14
View File
@@ -16,21 +16,21 @@ type RsyncStats struct {
}
type ProgressLine struct {
Phase string
Percent float64
Files int
Total int
SentBytes int64
Phase string
Percent float64
Files int
Total int
SentBytes int64
XferedBytes int64
}
type ProgressFields struct {
FileBytes int64
Pct int
SpeedBps int64
EtaSeconds int
XfrDone int
XfrTotal int
FileBytes int64 `json:"file_bytes"`
Pct int `json:"pct"`
SpeedBps int64 `json:"speed_bps"`
EtaSeconds int `json:"eta_seconds"`
XfrDone int `json:"xfr_done"`
XfrTotal int `json:"xfr_total"`
}
var (
@@ -138,9 +138,9 @@ func parseProgressFields(line string) *ProgressFields {
}
pf := &ProgressFields{
FileBytes: bytes,
Pct: pct,
SpeedBps: speedBps,
FileBytes: bytes,
Pct: pct,
SpeedBps: speedBps,
EtaSeconds: etaSecs,
}
internal/syncengine/queue.go
+46 -21
View File
@@ -13,60 +13,85 @@ type Queue struct {
}
type RunInfo struct {
JobID int64
Cancel func()
ByUser bool
JobID int64
SyncPairID int64
Cancel func()
CancelledBy bool
}
func NewQueue() *Queue {
return &Queue{runs: make(map[int64]*RunInfo)}
}
func (q *Queue) Enqueue(syncPairID, jobID int64, cancel func()) error {
func (q *Queue) Enqueue(syncPairID, jobID int64, cancelFn func()) error {
q.mu.Lock()
defer q.mu.Unlock()
if _, exists := q.runs[syncPairID]; exists {
if _, exists := q.runs[jobID]; exists {
return ErrAlreadyRunning
}
q.runs[syncPairID] = &RunInfo{JobID: jobID, Cancel: cancel}
for _, info := range q.runs {
if info.SyncPairID == syncPairID {
return ErrAlreadyRunning
}
}
q.runs[jobID] = &RunInfo{JobID: jobID, SyncPairID: syncPairID, Cancel: cancelFn}
return nil
}
func (q *Queue) Dequeue(syncPairID int64) {
func (q *Queue) Dequeue(jobID int64) {
q.mu.Lock()
defer q.mu.Unlock()
delete(q.runs, syncPairID)
delete(q.runs, jobID)
}
func (q *Queue) IsRunning(syncPairID int64) bool {
func (q *Queue) IsRunning(jobID int64) bool {
q.mu.Lock()
defer q.mu.Unlock()
_, exists := q.runs[syncPairID]
_, exists := q.runs[jobID]
return exists
}
func (q *Queue) GetJobID(syncPairID int64) (int64, bool) {
func (q *Queue) GetByPair(syncPairID int64) (jobID int64, exists bool) {
q.mu.Lock()
defer q.mu.Unlock()
info, exists := q.runs[syncPairID]
if !exists {
return 0, false
for _, info := range q.runs {
if info.SyncPairID == syncPairID {
return info.JobID, true
}
}
return info.JobID, true
return 0, false
}
func (q *Queue) Cancel(syncPairID int64, byUser bool) {
func (q *Queue) Cancel(jobID int64, byUser bool) bool {
q.mu.Lock()
defer q.mu.Unlock()
if info, exists := q.runs[syncPairID]; exists && info.Cancel != nil {
info.ByUser = byUser
if info, exists := q.runs[jobID]; exists && info.Cancel != nil {
info.CancelledBy = byUser
info.Cancel()
return true
}
return false
}
func (q *Queue) IsCancelledByUser(syncPairID int64) bool {
func (q *Queue) IsCancelledByUser(jobID int64) bool {
q.mu.Lock()
defer q.mu.Unlock()
info, exists := q.runs[syncPairID]
return exists && info.ByUser
info, exists := q.runs[jobID]
return exists && info.CancelledBy
}
func (q *Queue) RunningJobs() []int64 {
q.mu.Lock()
defer q.mu.Unlock()
ids := make([]int64, 0, len(q.runs))
for id := range q.runs {
ids = append(ids, id)
}
return ids
}
func (q *Queue) Len() int {
q.mu.Lock()
defer q.mu.Unlock()
return len(q.runs)
}
internal/syncengine/queue_test.go
+15 -13
View File
@@ -7,7 +7,7 @@ import (
func TestQueue(t *testing.T) {
q := NewQueue()
if q.IsRunning(1) {
if q.IsRunning(100) {
t.Error("queue should be empty")
}
@@ -16,35 +16,37 @@ func TestQueue(t *testing.T) {
err := q.Enqueue(1, 100, cancel)
if err != nil {
t.Errorf("Enqueue(1) unexpected error: %v", err)
t.Errorf("Enqueue(1, 100) unexpected error: %v", err)
}
if !q.IsRunning(1) {
t.Error("queue should contain syncPair 1")
if !q.IsRunning(100) {
t.Error("queue should contain job 100")
}
jobID, ok := q.GetJobID(1)
jobID, ok := q.GetByPair(1)
if !ok || jobID != 100 {
t.Errorf("GetJobID(1) = %d, %v, want 100, true", jobID, ok)
t.Errorf("GetByPair(1) = %d, %v, want 100, true", jobID, ok)
}
err = q.Enqueue(1, 200, nil)
if err != ErrAlreadyRunning {
t.Errorf("Enqueue(1) again = %v, want ErrAlreadyRunning", err)
t.Errorf("Enqueue(1, 200) = %v, want ErrAlreadyRunning", err)
}
q.Cancel(1, true)
ok = q.Cancel(100, true)
if !ok {
t.Error("Cancel(100) should return true")
}
if !cancelCalled {
t.Error("Cancel should have called the cancel func")
}
if !q.IsCancelledByUser(1) {
t.Error("IsCancelledByUser should return true after Cancel(1, true)")
if !q.IsCancelledByUser(100) {
t.Error("IsCancelledByUser should return true after Cancel(100, true)")
}
q.Dequeue(1)
q.Dequeue(100)
q.Dequeue(1)
if q.IsRunning(1) {
if q.IsRunning(100) {
t.Error("queue should be empty after Dequeue")
}
}
internal/syncengine/rsync_runner.go
+117 -7
View File
@@ -4,11 +4,97 @@ import (
"context"
"fmt"
"io"
"log/slog"
"os/exec"
"path/filepath"
"strconv"
"strings"
)
var allowedRsyncFlags = map[string]bool{
"-v": true,
"-vv": true,
"-q": true,
"-h": true,
"-P": true,
"-n": true,
"-z": true,
"-c": true,
"-u": true,
"-W": true,
"-i": true,
"-a": true,
"-r": true,
"-l": true,
"-t": true,
"-p": true,
"-g": true,
"-o": true,
"-D": true,
"--verbose": true,
"--quiet": true,
"--help": true,
"--partial": true,
"--partial-dir": true,
"--delay-updates": true,
"--delete": true,
"--delete-before": true,
"--delete-after": true,
"--delete-excluded": true,
"--exclude": true,
"--exclude-from": true,
"--dry-run": true,
"--compress": true,
"--skip-compress": true,
"--whole-file": true,
"--checksum": true,
"--update": true,
"--existing": true,
"--ignore-existing": true,
"--remove-source-files": true,
"--chmod": true,
"--owner": true,
"--group": true,
"--perms": true,
"--executability": true,
"--acls": true,
"--xattrs": true,
"--numeric-ids": true,
"--fake-super": true,
"--bwlimit": true,
"--max-size": true,
"--min-size": true,
"--append": true,
"--append-verify": true,
"--itemize-changes": true,
}
var blockedRsyncFlags = map[string]bool{
"--rsync-path": true,
"-e": true,
"--files-from": true,
"--read-batch": true,
"--write-batch": true,
"--log-file": true,
}
func isSafeRsyncFlag(flag string) bool {
if allowedRsyncFlags[flag] {
return true
}
safePrefixes := []string{
"-a", "-v", "-z", "-P", "-n", "-c", "-u", "-W", "-i",
"--exclude=", "--chmod=", "--bwlimit=", "--max-size=", "--min-size=",
"--partial-dir=", "--skip-compress=",
}
for _, p := range safePrefixes {
if strings.HasPrefix(flag, p) {
return true
}
}
return false
}
type RsyncResult struct {
ExitCode int
Stdout string
@@ -40,13 +126,13 @@ func NewRsyncRunner(sshDir, privKey string) *RsyncRunner {
}
func (r *RsyncRunner) Run(ctx context.Context, pair *SyncPairConfig, onLine func(stream string, line string)) (*RsyncResult, error) {
cmd := r.buildRsyncCmd(pair)
cmd := r.buildRsyncCmd(ctx, pair)
return r.runCmd(ctx, cmd, onLine)
}
func (r *RsyncRunner) buildRsyncCmd(pair *SyncPairConfig) *exec.Cmd {
func (r *RsyncRunner) buildRsyncCmd(ctx context.Context, pair *SyncPairConfig) *exec.Cmd {
args := r.buildArgs(pair)
cmd := exec.CommandContext(context.Background(), "rsync", args...)
cmd := exec.CommandContext(ctx, "rsync", args...)
if r.privKey != "" {
sshCmd := fmt.Sprintf("ssh -i %s -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile=%s",
r.privKey, strings.TrimRight(r.sshDir, "/")+"/known_hosts")
@@ -59,7 +145,19 @@ func (r *RsyncRunner) buildArgs(pair *SyncPairConfig) []string {
var args []string
flags := strings.Fields(pair.RsyncFlags)
args = append(args, flags...)
for _, flag := range flags {
if strings.HasPrefix(flag, "-") {
if blockedRsyncFlags[flag] {
slog.Warn("blocked dangerous rsync flag", "flag", flag)
continue
}
if !isSafeRsyncFlag(flag) {
slog.Warn("disallowed rsync flag", "flag", flag)
continue
}
}
args = append(args, flag)
}
for _, pattern := range pair.ExcludePatterns {
args = append(args, "--exclude="+pattern)
@@ -69,6 +167,7 @@ func (r *RsyncRunner) buildArgs(pair *SyncPairConfig) []string {
args = append(args, "--delete")
}
args = append(args, "--")
src := ensureDirSlash(pair.Source)
if pair.Direction == "pull" {
args = append(args, pair.Dest, src)
@@ -113,12 +212,23 @@ func (r *RsyncRunner) RunRemote(ctx context.Context, pair *SyncPairConfig, src *
innerSSH := fmt.Sprintf("ssh -i %s -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile=%s",
destKey, filepath.Join(r.sshDir, "known_hosts"))
rsyncFlags := strings.Join(args[:len(args)-2], " ")
rsyncFlags := args[:len(args)-2]
sourcePath := args[len(args)-2]
destPath := args[len(args)-1]
remoteCmd := fmt.Sprintf("rsync %s -e %q %s %s",
rsyncFlags, innerSSH, sourcePath, destPath)
var rsyncCmd []string
rsyncCmd = append(rsyncCmd, "rsync")
rsyncCmd = append(rsyncCmd, "-e")
rsyncCmd = append(rsyncCmd, innerSSH)
rsyncCmd = append(rsyncCmd, rsyncFlags...)
rsyncCmd = append(rsyncCmd, sourcePath, destPath)
remoteCmd := "rsync"
for _, arg := range rsyncFlags {
remoteCmd += " " + strconv.Quote(arg)
}
remoteCmd += " -e " + strconv.Quote(innerSSH) + " " + strconv.Quote(sourcePath) + " " + strconv.Quote(destPath)
remoteCmd = "sh -c " + strconv.Quote(remoteCmd)
sshArgs := []string{
"-i", src.PrivKey,
internal/syncengine/rsync_runner_test.go
+2 -1
View File
@@ -1,6 +1,7 @@
package syncengine
import (
"context"
"strings"
"testing"
)
@@ -281,7 +282,7 @@ func TestRun_FlagsPreservedWithPrivKey(t *testing.T) {
ExcludePatterns: []string{},
}
cmd := runner.buildRsyncCmd(pair)
cmd := runner.buildRsyncCmd(context.Background(), pair)
if cmd.Args[0] != "rsync" {
t.Errorf("cmd.Args[0] = %q, want 'rsync'", cmd.Args[0])