feat(protocol): job messages, protocol version and job model (#7)

2026-10-12 18:38:39 +02:00
parent eb41967cd1
commit 9e09de0a16
8 changed files with 314 additions and 16 deletions
@@ -2,7 +2,10 @@
 // open browser streams.
 package events

-import "sync"
+import (
+	"strconv"
+	"sync"
+)

 // Topic names what changed. Subscribers refetch that part of the UI.
 type Topic string
@@ -11,8 +14,13 @@ type Topic string
 const (
 	Services Topic = "services"
 	Agents   Topic = "agents"
+	Jobs     Topic = "jobs" // a job was created or changed status
 )

+// JobTopic is signalled when job id gets new output or changes status. The
+// name avoids ":" so it is safe inside hx-trigger="sse:job-42".
+func JobTopic(id int64) Topic { return Topic("job-" + strconv.FormatInt(id, 10)) }
+
 // Broker is an in-process publish/subscribe hub.
 type Broker struct {
 	mu   sync.Mutex
@@ -46,3 +46,7 @@ func TestBroker_SlowSubscriberDoesNotBlockPublish(t *testing.T) {
 		t.Fatal("Publish blocked on a slow subscriber")
 	}
 }
+
+func TestJobTopic(t *testing.T) {
+	assert.Equal(t, Topic("job-42"), JobTopic(42))
+}
@@ -0,0 +1,116 @@
+package model
+
+import (
+	"errors"
+	"time"
+)
+
+// Sentinel errors for jobs.
+var (
+	// ErrJobActive means the service already has a queued or running job.
+	// CreateJob and jobs.Service.Trigger return it together with that job.
+	ErrJobActive = errors.New("an update is already in progress for this service")
+	// ErrAgentOffline means no connected agent can run the job.
+	ErrAgentOffline = errors.New("agent offline")
+	// ErrAgentTooOld means the only agent that could run the job speaks a
+	// protocol version without jobs.
+	ErrAgentTooOld = errors.New("agent too old for updates")
+)
+
+// NotUpdatableError says why a service cannot be updated. Reason is shown
+// in the UI as is.
+type NotUpdatableError struct {
+	Reason string
+}
+
+func (e *NotUpdatableError) Error() string { return "not updatable: " + e.Reason }
+
+// Reasons a service cannot be updated.
+const (
+	ReasonSelf      = "updated manually outside the system"
+	ReasonContainer = "plain containers cannot be updated"
+	ReasonNoTag     = "follows no tag"
+	ReasonRemoved   = "service no longer exists"
+)
+
+// JobStatus is the state of an update job.
+type JobStatus string
+
+// Job statuses. A job moves queued -> running -> succeeded|failed, or
+// straight from queued to failed.
+const (
+	JobQueued    JobStatus = "queued"
+	JobRunning   JobStatus = "running"
+	JobSucceeded JobStatus = "succeeded"
+	JobFailed    JobStatus = "failed"
+)
+
+// JobTrigger says what created a job. Only TriggerManual is used in phase 2.
+type JobTrigger string
+
+// Job triggers.
+const (
+	TriggerManual  JobTrigger = "manual"
+	TriggerWebhook JobTrigger = "webhook"
+	TriggerPoll    JobTrigger = "poll"
+)
+
+// Job is one update of one service.
+type Job struct {
+	ID              int64
+	ServiceID       int64
+	ServiceName     string // Service.DisplayName() of the job's service
+	AgentID         *int64 // agent the job was sent to; nil if it never was
+	AgentName       string
+	FromRef         string
+	ToRef           string // "repo:tag" when created; "repo:tag@digest" after success
+	Trigger         JobTrigger
+	RequestedBy     string // OIDC subject for manual jobs
+	Status          JobStatus
+	Output          string // empty in ListJobs results
+	OutputTruncated bool
+	Error           string
+	CreatedAt       time.Time
+	StartedAt       *time.Time
+	FinishedAt      *time.Time
+}
+
+// Active reports whether the job is queued or running.
+func (j Job) Active() bool { return j.Status == JobQueued || j.Status == JobRunning }
+
+// NewJob is what CreateJob needs to insert a queued job.
+type NewJob struct {
+	ServiceID   int64
+	AgentID     *int64
+	FromRef     string
+	ToRef       string
+	Trigger     JobTrigger
+	RequestedBy string
+	CreatedAt   time.Time
+}
+
+// JobFinish is a job's outcome as reported by the agent.
+type JobFinish struct {
+	Status          JobStatus // JobSucceeded or JobFailed
+	Ref             string    // deployed "repo:tag@digest"; set on success
+	Error           string
+	Output          string
+	OutputTruncated bool
+	StartedAt       time.Time // agent clock
+	FinishedAt      time.Time // agent clock
+}
+
+// JobFilter selects jobs for ListJobs. ServiceID 0 means all services.
+// Limit 0 means 100.
+type JobFilter struct {
+	ServiceID int64
+	Limit     int
+}
+
+// Route is the hub's choice of agent connection for a job. ConnID identifies
+// the connection, so a job is never sent over a connection that replaced the
+// one chosen.
+type Route struct {
+	AgentID int64
+	ConnID  uint64
+}
@@ -41,10 +41,12 @@ type Service struct {
 	ComposeProject string
 	ComposeService string
 	WorkingDir     string
+	ConfigFiles    string // raw config_files label, paths joined by ","
 	SwarmClusterID string
 	SwarmService   string
 	ContainerName  string
 	ImageRepo      string
+	DesiredRef     string // set by the last successful job; "" if never updated
 	ActualRef      string
 	ActualDigest   string
 	Status         string
@@ -52,6 +54,7 @@ type Service struct {
 	UpdateMode     string
 	FirstSeenAt    time.Time
 	UpdatedAt      time.Time
+	RemovedAt      *time.Time
 }

 // DisplayName is the human-facing name of the service.
@@ -39,3 +39,20 @@ func TestDisplayName(t *testing.T) {
 		})
 	}
 }
+
+func TestJob_Active(t *testing.T) {
+	cases := []struct {
+		status JobStatus
+		want   bool
+	}{
+		{JobQueued, true},
+		{JobRunning, true},
+		{JobSucceeded, false},
+		{JobFailed, false},
+	}
+	for _, c := range cases {
+		t.Run(string(c.status), func(t *testing.T) {
+			assert.Equal(t, c.want, Job{Status: c.status}.Active())
+		})
+	}
+}
@@ -0,0 +1,27 @@
+ALTER TABLE services ADD COLUMN config_files text NOT NULL DEFAULT '';
+ALTER TABLE services ADD COLUMN desired_ref  text NOT NULL DEFAULT '';
+
+CREATE TABLE update_jobs (
+    id               bigserial PRIMARY KEY,
+    service_id       bigint NOT NULL REFERENCES services (id),
+    agent_id         bigint REFERENCES agents (id),
+    from_ref         text NOT NULL DEFAULT '',
+    to_ref           text NOT NULL,
+    trigger          text NOT NULL CHECK (trigger IN ('manual', 'webhook', 'poll')),
+    requested_by     text NOT NULL DEFAULT '',
+    status           text NOT NULL CHECK (status IN ('queued', 'running', 'succeeded', 'failed')),
+    output           text NOT NULL DEFAULT '',
+    output_truncated boolean NOT NULL DEFAULT false,
+    error            text NOT NULL DEFAULT '',
+    created_at       timestamptz NOT NULL,
+    started_at       timestamptz,
+    finished_at      timestamptz
+);
+
+-- One queued or running job per service. CreateJob relies on this index to
+-- turn a second concurrent click into ErrJobActive.
+CREATE UNIQUE INDEX update_jobs_one_active ON update_jobs (service_id)
+    WHERE status IN ('queued', 'running');
+CREATE INDEX update_jobs_by_service ON update_jobs (service_id, id DESC);
+CREATE INDEX update_jobs_active_by_agent ON update_jobs (agent_id)
+    WHERE status IN ('queued', 'running');
@@ -1,33 +1,66 @@
 // Package protocol defines the messages agents and the master exchange over
 // their WebSocket connection. Every message is an Envelope whose Payload is
 // one of the payload types below, selected by Type.
+//
+// Both sides close the connection on a message they cannot parse (bad JSON,
+// or a known Type whose payload does not decode). A message with an unknown
+// Type is logged and ignored, so a newer peer can add message types without
+// breaking an older one.
 package protocol

 import (
 	"encoding/json"
 	"fmt"
+	"time"
 )

+// Version is the protocol version this build speaks. Agents send it in
+// Hello. Phase-1 agents do not send the field, which reads as 0.
+const Version = 2
+
+// MinJobVersion is the lowest agent protocol version the master sends jobs to.
+const MinJobVersion = 2
+
+// MaxJobOutput caps the job output an agent keeps and the master stores, in
+// bytes. Only the tail is kept. With JSON escaping (worst case 6 bytes per
+// byte) a JobResult carrying this much output still fits the 1 MiB read limit.
+const MaxJobOutput = 128 << 10
+
 // Message types sent from agent to master.
 const (
-	TypeHello    = "hello"
-	TypeSnapshot = "snapshot"
-	TypeUpsert   = "upsert"
-	TypeRemove   = "remove"
+	TypeHello       = "hello"
+	TypeSnapshot    = "snapshot"
+	TypeUpsert      = "upsert"
+	TypeRemove      = "remove"
+	TypeJobProgress = "job.progress"
+	TypeJobResult   = "job.result"
 )

-// Envelope wraps every message on the wire.
+// Message types sent from master to agent.
+const (
+	TypeJobStart = "job.start"
+	TypeJobAck   = "job.ack"
+)
+
+// Envelope wraps every message on the wire. ID is the job ID (decimal) on
+// job.* messages and empty on inventory messages.
 type Envelope struct {
 	Type    string          `json:"type"`
+	ID      string          `json:"id,omitempty"`
 	Payload json.RawMessage `json:"payload,omitempty"`
 }

 // Hello is the first message an agent sends on every connection.
 type Hello struct {
-	AgentVersion   string `json:"agent_version"`
-	Hostname       string `json:"hostname"`
-	SwarmManager   bool   `json:"swarm_manager"`
-	SwarmClusterID string `json:"swarm_cluster_id,omitempty"`
+	AgentVersion    string `json:"agent_version"`
+	ProtocolVersion int    `json:"protocol_version,omitempty"`
+	Hostname        string `json:"hostname"`
+	SwarmManager    bool   `json:"swarm_manager"`
+	SwarmClusterID  string `json:"swarm_cluster_id,omitempty"`
+	// KnownJobs lists every job ID the agent holds: queued, running, or
+	// finished but not yet acknowledged. The master fails its queued and
+	// running jobs for this agent that are not listed ("lost").
+	KnownJobs []string `json:"known_jobs,omitempty"`
 }

 // Kind says how a service is deployed.
@@ -47,12 +80,16 @@ type Service struct {
 	ComposeProject string `json:"compose_project,omitempty"`
 	ComposeService string `json:"compose_service,omitempty"`
 	WorkingDir     string `json:"working_dir,omitempty"`
-	SwarmService   string `json:"swarm_service,omitempty"`
-	ContainerName  string `json:"container_name,omitempty"`
-	Image          string `json:"image"`
-	Digest         string `json:"digest,omitempty"`
-	Status         string `json:"status"`
-	Self           bool   `json:"self,omitempty"`
+	// ConfigFiles is the raw com.docker.compose.project.config_files label:
+	// absolute host paths joined by ",". Kept as a string so Service stays
+	// comparable with ==.
+	ConfigFiles   string `json:"config_files,omitempty"`
+	SwarmService  string `json:"swarm_service,omitempty"`
+	ContainerName string `json:"container_name,omitempty"`
+	Image         string `json:"image"`
+	Digest        string `json:"digest,omitempty"`
+	Status        string `json:"status"`
+	Self          bool   `json:"self,omitempty"`
 }

 // Key identifies the service within one agent's report. It is stable across
@@ -83,6 +120,50 @@ type Remove struct {
 	Key string `json:"key"`
 }

+// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag";
+// the agent resolves the tag's current digest and deploys "repo:tag@digest".
+// The agent reads the Compose working directory and config files from its
+// own Docker daemon, not from this message.
+type JobStart struct {
+	Kind           Kind   `json:"kind"`
+	ComposeProject string `json:"compose_project,omitempty"`
+	ComposeService string `json:"compose_service,omitempty"`
+	SwarmClusterID string `json:"swarm_cluster_id,omitempty"`
+	SwarmService   string `json:"swarm_service,omitempty"`
+	Ref            string `json:"ref"`
+}
+
+// JobProgress carries output produced since the previous JobProgress for the
+// same job. The first JobProgress of a job (Output may be empty) tells the
+// master the job is running.
+type JobProgress struct {
+	Output string `json:"output"`
+}
+
+// Job result statuses.
+const (
+	ResultSucceeded = "succeeded"
+	ResultFailed    = "failed"
+)
+
+// JobResult is a job's outcome. The agent keeps it until the master sends
+// JobAck for the same ID, and resends it after every reconnect until then.
+type JobResult struct {
+	Status string `json:"status"`        // ResultSucceeded or ResultFailed
+	Ref    string `json:"ref,omitempty"` // deployed "repo:tag@digest", set on success
+	Error  string `json:"error,omitempty"`
+	// Output is the job's complete output, tail-capped at MaxJobOutput. It
+	// replaces whatever the master assembled from JobProgress messages.
+	Output          string    `json:"output"`
+	OutputTruncated bool      `json:"output_truncated,omitempty"`
+	StartedAt       time.Time `json:"started_at"`
+	FinishedAt      time.Time `json:"finished_at"`
+}
+
+// JobAck tells the agent the master has stored the JobResult with the same
+// ID. It has no fields.
+type JobAck struct{}
+
 // Encode wraps payload in an Envelope of the given type.
 func Encode(typ string, payload any) (Envelope, error) {
 	raw, err := json.Marshal(payload)
@@ -92,6 +173,16 @@ func Encode(typ string, payload any) (Envelope, error) {
 	return Envelope{Type: typ, Payload: raw}, nil
 }

+// EncodeJob wraps a job message's payload in an Envelope carrying the job ID.
+func EncodeJob(typ, jobID string, payload any) (Envelope, error) {
+	env, err := Encode(typ, payload)
+	if err != nil {
+		return Envelope{}, err
+	}
+	env.ID = jobID
+	return env, nil
+}
+
 // Decode unpacks env's payload into T.
 func Decode[T any](env Envelope) (T, error) {
 	var v T
@@ -47,3 +47,35 @@ func TestDecodeRejectsMalformedPayload(t *testing.T) {
 	_, err := Decode[Hello](Envelope{Type: TypeHello, Payload: json.RawMessage(`{"hostname": 5}`)})
 	assert.ErrorContains(t, err, "decode hello")
 }
+
+func TestEncodeJob_CarriesID(t *testing.T) {
+	in := JobStart{Kind: KindSwarm, SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"}
+	env, err := EncodeJob(TypeJobStart, "42", in)
+	require.NoError(t, err)
+
+	raw, err := json.Marshal(env)
+	require.NoError(t, err)
+	var back Envelope
+	require.NoError(t, json.Unmarshal(raw, &back))
+
+	assert.Equal(t, "42", back.ID)
+	assert.Equal(t, TypeJobStart, back.Type)
+	out, err := Decode[JobStart](back)
+	require.NoError(t, err)
+	assert.Equal(t, in, out)
+}
+
+func TestHello_PhaseOneDecodesAsVersionZero(t *testing.T) {
+	var h Hello
+	require.NoError(t, json.Unmarshal([]byte(`{"agent_version":"x","hostname":"h","swarm_manager":false}`), &h))
+	assert.Equal(t, 0, h.ProtocolVersion)
+	assert.Nil(t, h.KnownJobs)
+}
+
+func TestEnvelope_InventoryHasNoID(t *testing.T) {
+	env, err := Encode(TypeSnapshot, Snapshot{})
+	require.NoError(t, err)
+	raw, err := json.Marshal(env)
+	require.NoError(t, err)
+	assert.NotContains(t, string(raw), `"id"`)
+}