internal/master/store/agents.go

@@ -64,7 +64,7 @@ func (s *Store) AgentByTokenHash(ctx context.Context, hash string) (model.Agent,
 }
 
 // RevokeAgent marks the agent revoked and its Compose and plain-container
-// services removed, in one transaction. Swarm rows are shared with other
+// services removed and its queued and running jobs failed, in one transaction. Swarm rows are shared with other
 // managers and left alone. Its token stops working immediately; closing a
 // live connection is the hub's job.
 func (s *Store) RevokeAgent(ctx context.Context, id int64, at time.Time) error {
@@ -83,6 +83,9 @@ func (s *Store) RevokeAgent(ctx context.Context, id int64, at time.Time) error {
 		 WHERE agent_id = $1 AND kind IN ('compose', 'container') AND removed_at IS NULL`, id, at); err != nil {
 		return fmt.Errorf("revoke agent: remove services: %w", err)
 	}
+	if _, err := tx.Exec(ctx, failActiveJobsSQL, id, "agent revoked", at); err != nil {
+		return fmt.Errorf("revoke agent: fail jobs: %w", err)
+	}
 	if err := tx.Commit(ctx); err != nil {
 		return fmt.Errorf("revoke agent: commit: %w", err)
 	}

internal/master/store/agents_integration_test.go

@@ -138,3 +138,24 @@ func TestRevokedAgent_UpsertDoesNotResurrectRemovedRow(t *testing.T) {
 	require.NoError(t, err)
 	assert.Empty(t, got, "the revoked agent's removed row must stay removed")
 }
+
+func TestRevokeAgent_FailsActiveJobs(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, svcA := jobFixture(t, s, "a")
+	b, svcB := jobFixture(t, s, "b")
+	ja, err := s.CreateJob(ctx, newJob(svcA, a))
+	require.NoError(t, err)
+	jb, err := s.CreateJob(ctx, newJob(svcB, b))
+	require.NoError(t, err)
+
+	require.NoError(t, s.RevokeAgent(ctx, a, t0))
+
+	got, err := s.GetJob(ctx, ja.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobFailed, got.Status)
+	assert.Equal(t, "agent revoked", got.Error)
+	got, err = s.GetJob(ctx, jb.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobQueued, got.Status)
+}

internal/master/store/jobs.go

@@ -0,0 +1,204 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+const lostJobMessage = "lost: the agent did not know this job when it reconnected"
+
+// failActiveJobsSQL fails every queued or running job of agent $1 with
+// message $2 at time $3.
+const failActiveJobsSQL = `
+UPDATE update_jobs SET status = 'failed', error = $2, finished_at = $3
+WHERE agent_id = $1 AND status IN ('queued', 'running')`
+
+const jobSelectSQL = `
+SELECT j.id, j.service_id, s.kind, s.compose_project, s.compose_service, s.swarm_service, s.container_name,
+       j.agent_id, COALESCE(a.name, ''), j.from_ref, j.to_ref, j.trigger, j.requested_by, j.status,
+       %s, j.output_truncated, j.error, j.created_at, j.started_at, j.finished_at
+FROM update_jobs j
+JOIN services s ON s.id = j.service_id
+LEFT JOIN agents a ON a.id = j.agent_id`
+
+func scanJob(row pgx.Row) (model.Job, error) {
+	var j model.Job
+	var svc model.Service
+	var kind, trigger, status string
+	err := row.Scan(&j.ID, &j.ServiceID, &kind, &svc.ComposeProject, &svc.ComposeService, &svc.SwarmService, &svc.ContainerName,
+		&j.AgentID, &j.AgentName, &j.FromRef, &j.ToRef, &trigger, &j.RequestedBy, &status,
+		&j.Output, &j.OutputTruncated, &j.Error, &j.CreatedAt, &j.StartedAt, &j.FinishedAt)
+	svc.Kind = protocol.Kind(kind)
+	j.ServiceName = svc.DisplayName()
+	j.Trigger = model.JobTrigger(trigger)
+	j.Status = model.JobStatus(status)
+	return j, err
+}
+
+func (s *Store) getJob(ctx context.Context, id int64) (model.Job, error) {
+	j, err := scanJob(s.pool.QueryRow(ctx, fmt.Sprintf(jobSelectSQL, "j.output")+` WHERE j.id = $1`, id))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return model.Job{}, model.ErrNotFound
+	}
+	if err != nil {
+		return model.Job{}, fmt.Errorf("get job: %w", err)
+	}
+	return j, nil
+}
+
+// CreateJob inserts a queued job. If the service already has a queued or
+// running job, it returns that job and model.ErrJobActive (detected via the
+// unique index, so it holds under concurrent calls).
+func (s *Store) CreateJob(ctx context.Context, j model.NewJob) (model.Job, error) {
+	var id int64
+	err := s.pool.QueryRow(ctx, `
+		INSERT INTO update_jobs (service_id, agent_id, from_ref, to_ref, trigger, requested_by, status, created_at)
+		VALUES ($1, $2, $3, $4, $5, $6, 'queued', $7)
+		RETURNING id`,
+		j.ServiceID, j.AgentID, j.FromRef, j.ToRef, string(j.Trigger), j.RequestedBy, j.CreatedAt).Scan(&id)
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == "23505" && pgErr.ConstraintName == "update_jobs_one_active" {
+		active, err := scanJob(s.pool.QueryRow(ctx, fmt.Sprintf(jobSelectSQL, "j.output")+
+			` WHERE j.service_id = $1 AND j.status IN ('queued', 'running')`, j.ServiceID))
+		if err != nil {
+			return model.Job{}, fmt.Errorf("find active job: %w", err)
+		}
+		return active, model.ErrJobActive
+	}
+	if err != nil {
+		return model.Job{}, fmt.Errorf("create job: %w", err)
+	}
+	return s.getJob(ctx, id)
+}
+
+// FailJob sets a queued or running job to failed with msg and finished_at =
+// at. Returns false (no error) if the job is no longer active.
+func (s *Store) FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error) {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE update_jobs SET status = 'failed', error = $2, finished_at = $3
+		WHERE id = $1 AND status IN ('queued', 'running')`, id, msg, at)
+	if err != nil {
+		return false, fmt.Errorf("fail job: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// GetJob returns one job with its output. Returns model.ErrNotFound if missing.
+func (s *Store) GetJob(ctx context.Context, id int64) (model.Job, error) {
+	return s.getJob(ctx, id)
+}
+
+// ListJobs returns jobs newest first, without Output.
+func (s *Store) ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error) {
+	limit := f.Limit
+	if limit <= 0 {
+		limit = 100
+	}
+	rows, err := s.pool.Query(ctx, fmt.Sprintf(jobSelectSQL, "''")+`
+		WHERE ($1 = 0 OR j.service_id = $1)
+		ORDER BY j.id DESC
+		LIMIT $2`, f.ServiceID, limit)
+	if err != nil {
+		return nil, fmt.Errorf("list jobs: %w", err)
+	}
+	defer rows.Close()
+	var out []model.Job
+	for rows.Next() {
+		j, err := scanJob(rows)
+		if err != nil {
+			return nil, fmt.Errorf("scan job: %w", err)
+		}
+		out = append(out, j)
+	}
+	return out, rows.Err()
+}
+
+// FailLostJobs fails every queued or running job sent to agentID whose ID is
+// not in known, with a "lost" error. Returns the IDs it failed.
+func (s *Store) FailLostJobs(ctx context.Context, agentID int64, known []int64, at time.Time) ([]int64, error) {
+	if known == nil {
+		known = []int64{}
+	}
+	rows, err := s.pool.Query(ctx, `
+		UPDATE update_jobs SET status = 'failed', error = $3, finished_at = $4
+		WHERE agent_id = $1 AND status IN ('queued', 'running') AND NOT (id = ANY($2))
+		RETURNING id`, agentID, known, lostJobMessage, at)
+	if err != nil {
+		return nil, fmt.Errorf("fail lost jobs: %w", err)
+	}
+	defer rows.Close()
+	var ids []int64
+	for rows.Next() {
+		var id int64
+		if err := rows.Scan(&id); err != nil {
+			return nil, fmt.Errorf("scan lost job: %w", err)
+		}
+		ids = append(ids, id)
+	}
+	return ids, rows.Err()
+}
+
+// JobProgress appends chunk to the output of a queued or running job sent to
+// agentID, keeping the last protocol.MaxJobOutput characters, and moves a
+// queued job to running (started_at = at if unset). Returns false if no such
+// active job belongs to agentID.
+func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk string, at time.Time) (bool, error) {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE update_jobs SET
+			status           = 'running',
+			started_at       = COALESCE(started_at, $4),
+			output_truncated = output_truncated OR length(output) + length($3::text) > $5,
+			output           = right(output || $3::text, $5)
+		WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running')`,
+		jobID, agentID, chunk, at, protocol.MaxJobOutput)
+	if err != nil {
+		return false, fmt.Errorf("job progress: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// FinishJob records the agent's result for a queued or running job sent to
+// agentID: status, error, output (replacing what progress assembled),
+// started_at (kept if already set), finished_at; to_ref = f.Ref when set. On
+// success it also sets the service's desired_ref = f.Ref, in the same
+// transaction. Returns false if no such active job belongs to agentID.
+func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error) {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return false, fmt.Errorf("finish job: begin: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	var serviceID int64
+	err = tx.QueryRow(ctx, `
+		UPDATE update_jobs SET
+			status = $3, error = $4, output = $5, output_truncated = $6,
+			to_ref = CASE WHEN $7::text <> '' THEN $7::text ELSE to_ref END,
+			started_at = COALESCE(started_at, $8), finished_at = $9
+		WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running')
+		RETURNING service_id`,
+		jobID, agentID, string(f.Status), f.Error, f.Output, f.OutputTruncated, f.Ref, f.StartedAt, f.FinishedAt).Scan(&serviceID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("finish job: %w", err)
+	}
+	if f.Status == model.JobSucceeded {
+		if _, err := tx.Exec(ctx, `UPDATE services SET desired_ref = $2 WHERE id = $1`, serviceID, f.Ref); err != nil {
+			return false, fmt.Errorf("finish job: desired ref: %w", err)
+		}
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return false, fmt.Errorf("finish job: commit: %w", err)
+	}
+	return true, nil
+}

internal/master/store/jobs_integration_test.go

@@ -0,0 +1,305 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// jobFixture seeds one agent with one Compose service and returns their IDs.
+func jobFixture(t *testing.T, s *Store, agentName string) (agentID, serviceID int64) {
+	t.Helper()
+	ctx := context.Background()
+	agentID = newAgent(t, s, agentName)
+	require.NoError(t, s.ApplySnapshot(ctx, agentID, worker, []protocol.Service{slusk}, t0))
+	svcs, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	for _, svc := range svcs {
+		if svc.AgentID != nil && *svc.AgentID == agentID {
+			return agentID, svc.ID
+		}
+	}
+	t.Fatal("service not found")
+	return
+}
+
+func newJob(serviceID, agentID int64) model.NewJob {
+	return model.NewJob{
+		ServiceID: serviceID, AgentID: &agentID,
+		FromRef: "r:1@sha256:a", ToRef: "r:1", Trigger: model.TriggerManual,
+		RequestedBy: "sub", CreatedAt: t0,
+	}
+}
+
+func TestCreateJob_OneActivePerService(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	agentID, svcID := jobFixture(t, s, "a")
+
+	var wg sync.WaitGroup
+	jobs := make([]model.Job, 2)
+	errs := make([]error, 2)
+	for i := range jobs {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			jobs[i], errs[i] = s.CreateJob(ctx, newJob(svcID, agentID))
+		}()
+	}
+	wg.Wait()
+
+	assert.Equal(t, jobs[0].ID, jobs[1].ID)
+	var failures int
+	for _, err := range errs {
+		if err != nil {
+			assert.ErrorIs(t, err, model.ErrJobActive)
+			failures++
+		}
+	}
+	assert.Equal(t, 1, failures)
+
+	ok, err := s.FailJob(ctx, jobs[0].ID, "x", t0)
+	require.NoError(t, err)
+	assert.True(t, ok)
+	ok, err = s.FailJob(ctx, jobs[0].ID, "x", t0)
+	require.NoError(t, err)
+	assert.False(t, ok)
+
+	next, err := s.CreateJob(ctx, newJob(svcID, agentID))
+	require.NoError(t, err)
+	assert.NotEqual(t, jobs[0].ID, next.ID)
+}
+
+func TestFailLostJobs(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, svcA := jobFixture(t, s, "a")
+	b := newAgent(t, s, "b")
+	require.NoError(t, s.ApplySnapshot(ctx, b, worker, []protocol.Service{adhoc}, t0))
+	all, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	var svcB int64
+	for _, svc := range all {
+		if svc.ID != svcA {
+			svcB = svc.ID
+		}
+	}
+
+	// Only one active job per service, so A's jobs go through the lifecycle.
+	j3, err := s.CreateJob(ctx, newJob(svcA, a))
+	require.NoError(t, err)
+	ok, err := s.FinishJob(ctx, a, j3.ID, model.JobFinish{Status: model.JobSucceeded, Ref: "r:1@sha256:d", StartedAt: t0, FinishedAt: t0})
+	require.NoError(t, err)
+	require.True(t, ok)
+	j1, err := s.CreateJob(ctx, newJob(svcA, a))
+	require.NoError(t, err)
+	j4, err := s.CreateJob(ctx, newJob(svcB, b))
+	require.NoError(t, err)
+	_, err = s.JobProgress(ctx, b, j4.ID, "x", t0)
+	require.NoError(t, err)
+
+	ids, err := s.FailLostJobs(ctx, a, []int64{j3.ID}, t0)
+	require.NoError(t, err)
+	assert.Equal(t, []int64{j1.ID}, ids)
+
+	got, err := s.GetJob(ctx, j1.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobFailed, got.Status)
+	assert.Contains(t, got.Error, "lost")
+	got, err = s.GetJob(ctx, j3.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobSucceeded, got.Status)
+	got, err = s.GetJob(ctx, j4.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobRunning, got.Status)
+
+	// A running job the agent still knows is kept.
+	j5, err := s.CreateJob(ctx, newJob(svcA, a))
+	require.NoError(t, err)
+	_, err = s.JobProgress(ctx, a, j5.ID, "", t0)
+	require.NoError(t, err)
+	ids, err = s.FailLostJobs(ctx, a, []int64{j5.ID}, t0)
+	require.NoError(t, err)
+	assert.Empty(t, ids)
+}
+
+func TestJobProgress_OwnerOnly(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, svc := jobFixture(t, s, "a")
+	b := newAgent(t, s, "b")
+	j, err := s.CreateJob(ctx, newJob(svc, a))
+	require.NoError(t, err)
+
+	started := t0.Add(time.Second)
+	ok, err := s.JobProgress(ctx, a, j.ID, "hello ", started)
+	require.NoError(t, err)
+	assert.True(t, ok)
+	got, err := s.GetJob(ctx, j.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobRunning, got.Status)
+	require.NotNil(t, got.StartedAt)
+	assert.True(t, started.Equal(*got.StartedAt))
+
+	ok, err = s.JobProgress(ctx, b, j.ID, "evil", t0.Add(time.Minute))
+	require.NoError(t, err)
+	assert.False(t, ok)
+	got, err = s.GetJob(ctx, j.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "hello ", got.Output)
+
+	ok, err = s.JobProgress(ctx, a, j.ID, "world", t0.Add(time.Minute))
+	require.NoError(t, err)
+	assert.True(t, ok)
+	got, err = s.GetJob(ctx, j.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "hello world", got.Output)
+	assert.True(t, started.Equal(*got.StartedAt))
+
+	_, err = s.FailJob(ctx, j.ID, "x", t0)
+	require.NoError(t, err)
+	ok, err = s.JobProgress(ctx, a, j.ID, "late", t0)
+	require.NoError(t, err)
+	assert.False(t, ok)
+}
+
+func TestJobProgress_KeepsTail(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, svc := jobFixture(t, s, "a")
+	j, err := s.CreateJob(ctx, newJob(svc, a))
+	require.NoError(t, err)
+
+	_, err = s.JobProgress(ctx, a, j.ID, strings.Repeat("a", protocol.MaxJobOutput), t0)
+	require.NoError(t, err)
+	got, err := s.GetJob(ctx, j.ID)
+	require.NoError(t, err)
+	assert.False(t, got.OutputTruncated)
+
+	_, err = s.JobProgress(ctx, a, j.ID, "TAIL", t0)
+	require.NoError(t, err)
+	got, err = s.GetJob(ctx, j.ID)
+	require.NoError(t, err)
+	assert.True(t, got.OutputTruncated)
+	assert.Len(t, got.Output, protocol.MaxJobOutput)
+	assert.True(t, strings.HasSuffix(got.Output, "aaTAIL"))
+}
+
+func TestFinishJob_OnlyOnceAndOnlyByOwner(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, svc := jobFixture(t, s, "a")
+	b := newAgent(t, s, "b")
+	j, err := s.CreateJob(ctx, newJob(svc, a))
+	require.NoError(t, err)
+	_, err = s.JobProgress(ctx, a, j.ID, "partial", t0)
+	require.NoError(t, err)
+
+	fin := model.JobFinish{Status: model.JobSucceeded, Ref: "r:1@sha256:d", Output: "full", StartedAt: t0.Add(time.Hour), FinishedAt: t0.Add(2 * time.Hour)}
+	ok, err := s.FinishJob(ctx, b, j.ID, fin)
+	require.NoError(t, err)
+	assert.False(t, ok)
+
+	ok, err = s.FinishJob(ctx, a, j.ID, fin)
+	require.NoError(t, err)
+	assert.True(t, ok)
+	got, err := s.GetJob(ctx, j.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobSucceeded, got.Status)
+	assert.Equal(t, "r:1@sha256:d", got.ToRef)
+	assert.Equal(t, "full", got.Output)
+	assert.True(t, t0.Equal(*got.StartedAt), "started_at set by the first progress is kept")
+	assert.True(t, fin.FinishedAt.Equal(*got.FinishedAt))
+	service, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, "r:1@sha256:d", service.DesiredRef)
+
+	again := fin
+	again.Ref = "r:2@sha256:e"
+	again.Output = "other"
+	ok, err = s.FinishJob(ctx, a, j.ID, again)
+	require.NoError(t, err)
+	assert.False(t, ok)
+	got, err = s.GetJob(ctx, j.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "full", got.Output)
+	service, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, "r:1@sha256:d", service.DesiredRef)
+
+	j2, err := s.CreateJob(ctx, newJob(svc, a))
+	require.NoError(t, err)
+	ok, err = s.FinishJob(ctx, a, j2.ID, model.JobFinish{Status: model.JobFailed, Error: "boom", Output: "o", StartedAt: t0, FinishedAt: t0})
+	require.NoError(t, err)
+	assert.True(t, ok)
+	got, err = s.GetJob(ctx, j2.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.JobFailed, got.Status)
+	assert.Equal(t, "boom", got.Error)
+	assert.Equal(t, "r:1", got.ToRef)
+	service, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, "r:1@sha256:d", service.DesiredRef)
+}
+
+func TestGetJobAndListJobs(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, svcA := jobFixture(t, s, "a")
+	b := newAgent(t, s, "b")
+	require.NoError(t, s.ApplySnapshot(ctx, b, worker, []protocol.Service{adhoc}, t0))
+	all, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	var svcB int64
+	for _, svc := range all {
+		if svc.ID != svcA {
+			svcB = svc.ID
+		}
+	}
+
+	j1, err := s.CreateJob(ctx, newJob(svcA, a))
+	require.NoError(t, err)
+	_, err = s.JobProgress(ctx, a, j1.ID, "out", t0)
+	require.NoError(t, err)
+	_, err = s.FailJob(ctx, j1.ID, "x", t0)
+	require.NoError(t, err)
+	j2, err := s.CreateJob(ctx, newJob(svcA, a))
+	require.NoError(t, err)
+	j3, err := s.CreateJob(ctx, newJob(svcB, b))
+	require.NoError(t, err)
+
+	got, err := s.GetJob(ctx, j1.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "slusk/web", got.ServiceName)
+	assert.Equal(t, "a", got.AgentName)
+	assert.Equal(t, "out", got.Output)
+
+	_, err = s.GetJob(ctx, 99999)
+	assert.ErrorIs(t, err, model.ErrNotFound)
+
+	list, err := s.ListJobs(ctx, model.JobFilter{})
+	require.NoError(t, err)
+	require.Len(t, list, 3)
+	assert.Equal(t, []int64{j3.ID, j2.ID, j1.ID}, []int64{list[0].ID, list[1].ID, list[2].ID})
+	for _, j := range list {
+		assert.Empty(t, j.Output)
+	}
+
+	list, err = s.ListJobs(ctx, model.JobFilter{ServiceID: svcA})
+	require.NoError(t, err)
+	assert.Len(t, list, 2)
+
+	list, err = s.ListJobs(ctx, model.JobFilter{Limit: 1})
+	require.NoError(t, err)
+	assert.Len(t, list, 1)
+}

internal/master/store/services.go

@@ -39,13 +39,14 @@ func lockActiveAgent(ctx context.Context, tx pgx.Tx, agentID int64) error {
 
 const upsertServiceSQL = `
 INSERT INTO services (
-	kind, location_key, agent_id, compose_project, compose_service, working_dir,
+	kind, location_key, agent_id, compose_project, compose_service, working_dir, config_files,
 	swarm_cluster_id, swarm_service, container_name, image_repo, actual_ref, actual_digest,
 	status, self, first_seen_at, updated_at)
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $15)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $16)
 ON CONFLICT (location_key) DO UPDATE SET
 	agent_id      = EXCLUDED.agent_id,
 	working_dir   = EXCLUDED.working_dir,
+	config_files  = EXCLUDED.config_files,
 	image_repo    = EXCLUDED.image_repo,
 	actual_ref    = EXCLUDED.actual_ref,
 	actual_digest = EXCLUDED.actual_digest,
@@ -61,7 +62,7 @@ func upsertService(ctx context.Context, db execer, agentID int64, clusterID stri
 	}
 	_, err := db.Exec(ctx, upsertServiceSQL,
 		string(svc.Kind), model.LocationKey(agentID, clusterID, svc.Key()), agentID,
-		svc.ComposeProject, svc.ComposeService, svc.WorkingDir,
+		svc.ComposeProject, svc.ComposeService, svc.WorkingDir, svc.ConfigFiles,
 		clusterCol, svc.SwarmService, svc.ContainerName,
 		imageref.Parse(svc.Image).Repo, svc.Image, svc.Digest,
 		svc.Status, svc.Self, at)
@@ -144,14 +145,27 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age
 	return nil
 }
 
+const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''),
+	s.compose_project, s.compose_service, s.working_dir, s.config_files, s.swarm_cluster_id,
+	s.swarm_service, s.container_name, s.image_repo, s.desired_ref, s.actual_ref, s.actual_digest,
+	s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at, s.removed_at`
+
+func scanService(row pgx.Row) (model.Service, error) {
+	var m model.Service
+	var kind string
+	err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName,
+		&m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.SwarmClusterID,
+		&m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.DesiredRef, &m.ActualRef, &m.ActualDigest,
+		&m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
+	m.Kind = protocol.Kind(kind)
+	return m, err
+}
+
 // ListServices returns all services not marked removed: machine-bound
 // services grouped by agent name first, Swarm services last.
 func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) {
 	rows, err := s.pool.Query(ctx, `
-		SELECT s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''),
-		       s.compose_project, s.compose_service, s.working_dir, s.swarm_cluster_id,
-		       s.swarm_service, s.container_name, s.image_repo, s.actual_ref, s.actual_digest,
-		       s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at
+		SELECT `+serviceColumns+`
 		FROM services s LEFT JOIN agents a ON a.id = s.agent_id
 		WHERE s.removed_at IS NULL
 		ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`)
@@ -161,16 +175,27 @@ func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) {
 	defer rows.Close()
 	var out []model.Service
 	for rows.Next() {
-		var m model.Service
-		var kind string
-		if err := rows.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName,
-			&m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.SwarmClusterID,
-			&m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.ActualRef, &m.ActualDigest,
-			&m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt); err != nil {
+		m, err := scanService(rows)
+		if err != nil {
 			return nil, fmt.Errorf("scan service: %w", err)
 		}
-		m.Kind = protocol.Kind(kind)
 		out = append(out, m)
 	}
 	return out, rows.Err()
 }
+
+// GetService returns one service, including a removed one (RemovedAt set).
+// Returns model.ErrNotFound if the ID does not exist.
+func (s *Store) GetService(ctx context.Context, id int64) (model.Service, error) {
+	m, err := scanService(s.pool.QueryRow(ctx, `
+		SELECT `+serviceColumns+`
+		FROM services s LEFT JOIN agents a ON a.id = s.agent_id
+		WHERE s.id = $1`, id))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return model.Service{}, model.ErrNotFound
+	}
+	if err != nil {
+		return model.Service{}, fmt.Errorf("get service: %w", err)
+	}
+	return m, nil
+}

internal/master/store/services_integration_test.go

@@ -160,3 +160,36 @@ func TestListServices_SwarmLast(t *testing.T) {
 }
 
 func itoa(n int64) string { return strconv.FormatInt(n, 10) }
+
+func TestGetService_IncludesRemoved(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "node-2")
+	require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk}, t0))
+	svcs, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	require.Len(t, svcs, 1)
+
+	require.NoError(t, s.RemoveService(ctx, id, "", slusk.Key(), t0))
+	got, err := s.GetService(ctx, svcs[0].ID)
+	require.NoError(t, err)
+	assert.NotNil(t, got.RemovedAt)
+	assert.Equal(t, "node-2", got.AgentName)
+
+	_, err = s.GetService(ctx, 99999)
+	assert.ErrorIs(t, err, model.ErrNotFound)
+}
+
+func TestApplySnapshot_StoresConfigFiles(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "node-2")
+	withFiles := slusk
+	withFiles.ConfigFiles = "/srv/slusk/compose.yaml,/srv/slusk/compose.override.yaml"
+	require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{withFiles}, t0))
+
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, withFiles.ConfigFiles, got[0].ConfigFiles)
+}