darroyo
84b185be39
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
98 lines
1.8 KiB
Go
98 lines
1.8 KiB
Go
package syncengine
|
|
|
|
import (
|
|
"errors"
|
|
"sync"
|
|
)
|
|
|
|
var ErrAlreadyRunning = errors.New("job already running for this sync pair")
|
|
|
|
type Queue struct {
|
|
mu sync.Mutex
|
|
runs map[int64]*RunInfo
|
|
}
|
|
|
|
type RunInfo struct {
|
|
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, cancelFn func()) error {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if _, exists := q.runs[jobID]; exists {
|
|
return ErrAlreadyRunning
|
|
}
|
|
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(jobID int64) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
delete(q.runs, jobID)
|
|
}
|
|
|
|
func (q *Queue) IsRunning(jobID int64) bool {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
_, exists := q.runs[jobID]
|
|
return exists
|
|
}
|
|
|
|
func (q *Queue) GetByPair(syncPairID int64) (jobID int64, exists bool) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
for _, info := range q.runs {
|
|
if info.SyncPairID == syncPairID {
|
|
return info.JobID, true
|
|
}
|
|
}
|
|
return 0, false
|
|
}
|
|
|
|
func (q *Queue) Cancel(jobID int64, byUser bool) bool {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if info, exists := q.runs[jobID]; exists && info.Cancel != nil {
|
|
info.CancelledBy = byUser
|
|
info.Cancel()
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (q *Queue) IsCancelledByUser(jobID int64) bool {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
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)
|
|
}
|