shcizo/package-updater
1
0
Fork 0
Code Issues 8 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(store): available ref, update mode and services-by-repo for the webhook (#8)

Browse Source
This commit is contained in:
shcizo
2026-10-12 20:40:04 +02:00
parent 417ef315c2
commit cfc6ea6713
6 changed files with 240 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/master/store/jobs.go
+12 -4
View File
@@ -187,7 +187,9 @@ func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk str
// agentID: status, error, output (replacing what progress assembled),
// started_at (the agent's, or the existing one if the agent sent none),
// 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
// service's desired_ref = f.Ref and clears its available ref if that was
// reported no later than the job was created (the job then deployed the
// newest version), 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)
@@ -203,14 +205,15 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job
startedAt = &f.StartedAt
}
var serviceID int64
var createdAt time.Time
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($8, started_at), 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, startedAt, f.FinishedAt).Scan(&serviceID)
RETURNING service_id, created_at`,
jobID, agentID, string(f.Status), f.Error, f.Output, f.OutputTruncated, f.Ref, startedAt, f.FinishedAt).Scan(&serviceID, &createdAt)
if errors.Is(err, pgx.ErrNoRows) {
return false, nil
}
@@ -218,7 +221,12 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job
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 {
if _, err := tx.Exec(ctx, `
UPDATE services SET desired_ref = $2,
available_ref = CASE WHEN available_at <= $3 THEN '' ELSE available_ref END,
available_source = CASE WHEN available_at <= $3 THEN '' ELSE available_source END,
available_at = CASE WHEN available_at <= $3 THEN NULL ELSE available_at END
WHERE id = $1`, serviceID, f.Ref, createdAt); err != nil {
return false, fmt.Errorf("finish job: desired ref: %w", err)
}
}
internal/master/store/migrations/0003_webhook.sql
+6
View File
@@ -0,0 +1,6 @@
ALTER TABLE services ADD COLUMN available_ref text NOT NULL DEFAULT '';
ALTER TABLE services ADD COLUMN available_source text NOT NULL DEFAULT ''
CHECK (available_source IN ('', 'webhook', 'poll'));
ALTER TABLE services ADD COLUMN available_at timestamptz;
CREATE INDEX services_by_image_repo ON services (image_repo) WHERE removed_at IS NULL;
internal/master/store/services.go
+43 -5
View File
@@ -149,7 +149,8 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age
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.built, 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`
s.status, s.self, s.update_mode, s.available_ref, s.available_source, s.available_at,
s.first_seen_at, s.updated_at, s.removed_at`
func scanService(row pgx.Row) (model.Service, error) {
var m model.Service
@@ -157,7 +158,8 @@ func scanService(row pgx.Row) (model.Service, error) {
err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName,
&m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.Built, &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.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableSource, &m.AvailableAt,
&m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
m.Kind = protocol.Kind(kind)
return m, err
}
@@ -165,13 +167,23 @@ func scanService(row pgx.Row) (model.Service, error) {
// 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) {
return s.queryServices(ctx, "list services", "", nil)
}
// ServicesByRepo returns the services not marked removed whose image repo
// equals repo (case-sensitive), in the same order as ListServices.
func (s *Store) ServicesByRepo(ctx context.Context, repo string) ([]model.Service, error) {
return s.queryServices(ctx, "services by repo", "AND s.image_repo = $1", []any{repo})
}
func (s *Store) queryServices(ctx context.Context, what, extraWhere string, args []any) ([]model.Service, error) {
rows, err := s.pool.Query(ctx, `
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`)
WHERE s.removed_at IS NULL `+extraWhere+`
ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`, args...)
if err != nil {
return nil, fmt.Errorf("list services: %w", err)
return nil, fmt.Errorf("%s: %w", what, err)
}
defer rows.Close()
var out []model.Service
@@ -185,6 +197,32 @@ func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) {
return out, rows.Err()
}
// SetAvailable records a newer ref for the service, replacing any earlier
// suggestion.
func (s *Store) SetAvailable(ctx context.Context, id int64, a model.Available) error {
if _, err := s.pool.Exec(ctx, `
UPDATE services SET available_ref = $2, available_source = $3, available_at = $4
WHERE id = $1`, id, a.Ref, a.Source, a.At); err != nil {
return fmt.Errorf("set available: %w", err)
}
return nil
}
// SetUpdateMode sets the service's update mode (model.ModeManual or
// model.ModeAuto). Returns model.ErrNotFound if the service does not exist
// or is marked removed.
func (s *Store) SetUpdateMode(ctx context.Context, id int64, mode string) error {
tag, err := s.pool.Exec(ctx,
`UPDATE services SET update_mode = $2 WHERE id = $1 AND removed_at IS NULL`, id, mode)
if err != nil {
return fmt.Errorf("set update mode: %w", err)
}
if tag.RowsAffected() == 0 {
return model.ErrNotFound
}
return nil
}
// 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) {
internal/master/store/webhook_integration_test.go
+149
View File
@@ -0,0 +1,149 @@
//go:build integration
package store
import (
"context"
"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"
)
func TestServicesByRepo(t *testing.T) {
s := openTestStore(t)
ctx := context.Background()
id := newAgent(t, s, "node-2")
other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "o", ComposeService: "o", Image: "gitea.shcizo.se/shcizo/Slusk:1", Status: "running"}
gone := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "g", ComposeService: "g", Image: "gitea.shcizo.se/shcizo/slusk:0.9", Status: "running"}
require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk, other, gone, adhoc}, t0))
require.NoError(t, s.RemoveService(ctx, id, "", gone.Key(), t0))
got, err := s.ServicesByRepo(ctx, "gitea.shcizo.se/shcizo/slusk")
require.NoError(t, err)
assert.Equal(t, []string{"slusk/web"}, names(got), "removed rows and rows of other (case-sensitive) repos are excluded")
assert.Equal(t, "node-2", got[0].AgentName)
none, err := s.ServicesByRepo(ctx, "nope")
require.NoError(t, err)
assert.Empty(t, none)
}
func TestSetAvailable_RoundTrip(t *testing.T) {
s := openTestStore(t)
ctx := context.Background()
_, svc := jobFixture(t, s, "a")
require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "r:2", Source: model.SourceWebhook, At: t0}))
got, err := s.GetService(ctx, svc)
require.NoError(t, err)
assert.Equal(t, "r:2", got.AvailableRef)
assert.Equal(t, model.SourceWebhook, got.AvailableSource)
require.NotNil(t, got.AvailableAt)
assert.True(t, t0.Equal(*got.AvailableAt))
}
func TestSetUpdateMode(t *testing.T) {
s := openTestStore(t)
ctx := context.Background()
agent, svc := jobFixture(t, s, "a")
require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto))
got, err := s.GetService(ctx, svc)
require.NoError(t, err)
assert.Equal(t, model.ModeAuto, got.UpdateMode)
require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeManual))
got, _ = s.GetService(ctx, svc)
assert.Equal(t, model.ModeManual, got.UpdateMode)
assert.ErrorIs(t, s.SetUpdateMode(ctx, 99999, model.ModeAuto), model.ErrNotFound)
require.NoError(t, s.RemoveService(ctx, agent, "", slusk.Key(), t0))
assert.ErrorIs(t, s.SetUpdateMode(ctx, svc, model.ModeAuto), model.ErrNotFound)
}
func TestUpsertKeepsUpdateMode(t *testing.T) {
s := openTestStore(t)
ctx := context.Background()
agent, svc := jobFixture(t, s, "a")
got, err := s.GetService(ctx, svc)
require.NoError(t, err)
assert.Equal(t, model.ModeManual, got.UpdateMode, "a new service starts manual")
require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto))
changed := slusk
changed.Image = "gitea.shcizo.se/shcizo/slusk:1.5.0"
require.NoError(t, s.UpsertService(ctx, agent, "", changed, t0.Add(time.Minute)))
require.NoError(t, s.ApplySnapshot(ctx, agent, worker, []protocol.Service{changed}, t0.Add(2*time.Minute)))
got, err = s.GetService(ctx, svc)
require.NoError(t, err)
assert.Equal(t, model.ModeAuto, got.UpdateMode)
}
func TestFinishJob_ClearsAvailableOnlyWhenJobIsNewer(t *testing.T) {
succeed := model.JobFinish{Status: model.JobSucceeded, Ref: "r:2@sha256:d", StartedAt: t0, FinishedAt: t0}
tests := []struct {
name string
availableAt time.Time // job is created at t0
finish model.JobFinish
wantCleared bool
}{
{"suggestion before the job", t0.Add(-time.Minute), succeed, true},
{"suggestion at the same instant", t0, succeed, true},
{"suggestion after the job was created", t0.Add(time.Minute), succeed, false},
{"failed job keeps the suggestion", t0.Add(-time.Minute),
model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0}, false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
s := openTestStore(t)
ctx := context.Background()
agent, svc := jobFixture(t, s, "a")
require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "r:2", Source: model.SourceWebhook, At: tt.availableAt}))
j, err := s.CreateJob(ctx, newJob(svc, agent))
require.NoError(t, err)
ok, err := s.FinishJob(ctx, agent, j.ID, tt.finish)
require.NoError(t, err)
require.True(t, ok)
got, err := s.GetService(ctx, svc)
require.NoError(t, err)
if tt.wantCleared {
assert.Empty(t, got.AvailableRef)
assert.Empty(t, got.AvailableSource)
assert.Nil(t, got.AvailableAt)
} else {
assert.Equal(t, "r:2", got.AvailableRef)
assert.Equal(t, model.SourceWebhook, got.AvailableSource)
assert.NotNil(t, got.AvailableAt)
}
})
}
}
func TestFinishJob_WithoutAvailableLeavesItEmpty(t *testing.T) {
s := openTestStore(t)
ctx := context.Background()
agent, svc := jobFixture(t, s, "a")
j, err := s.CreateJob(ctx, newJob(svc, agent))
require.NoError(t, err)
ok, err := s.FinishJob(ctx, agent, j.ID, model.JobFinish{Status: model.JobSucceeded, Ref: "r:1@sha256:d", StartedAt: t0, FinishedAt: t0})
require.NoError(t, err)
require.True(t, ok)
got, err := s.GetService(ctx, svc)
require.NoError(t, err)
assert.Empty(t, got.AvailableRef)
assert.Nil(t, got.AvailableAt)
assert.Equal(t, "r:1@sha256:d", got.DesiredRef)
}