feat(agent): diff inventories and debounce docker events

internal/agent/inventory/diff.go

@@ -0,0 +1,25 @@
+package inventory
+
+import "github.com/shcizo/package-updater/internal/protocol"
+
+// Diff compares two snapshots. upserts holds services that are new or
+// changed in next; removes holds the keys of services missing from next.
+func Diff(prev, next []protocol.Service) (upserts []protocol.Service, removes []string) {
+	old := make(map[string]protocol.Service, len(prev))
+	for _, s := range prev {
+		old[s.Key()] = s
+	}
+	seen := make(map[string]struct{}, len(next))
+	for _, s := range next {
+		seen[s.Key()] = struct{}{}
+		if o, ok := old[s.Key()]; !ok || o != s {
+			upserts = append(upserts, s)
+		}
+	}
+	for _, s := range prev {
+		if _, ok := seen[s.Key()]; !ok {
+			removes = append(removes, s.Key())
+		}
+	}
+	return upserts, removes
+}

internal/agent/inventory/diff_test.go

@@ -0,0 +1,37 @@
+package inventory
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+func TestDiff(t *testing.T) {
+	a := protocol.Service{Kind: protocol.KindContainer, ContainerName: "a", Image: "a:1", Status: "running"}
+	b := protocol.Service{Kind: protocol.KindContainer, ContainerName: "b", Image: "b:1", Status: "running"}
+	bStopped := b
+	bStopped.Status = "stopped"
+	c := protocol.Service{Kind: protocol.KindContainer, ContainerName: "c", Image: "c:1", Status: "running"}
+
+	cases := []struct {
+		name        string
+		prev, next  []protocol.Service
+		wantUpserts []protocol.Service
+		wantRemoves []string
+	}{
+		{"unchanged", []protocol.Service{a, b}, []protocol.Service{a, b}, nil, nil},
+		{"added", []protocol.Service{a}, []protocol.Service{a, c}, []protocol.Service{c}, nil},
+		{"changed", []protocol.Service{a, b}, []protocol.Service{a, bStopped}, []protocol.Service{bStopped}, nil},
+		{"removed", []protocol.Service{a, b}, []protocol.Service{a}, nil, []string{"container|b"}},
+		{"from empty", nil, []protocol.Service{a}, []protocol.Service{a}, nil},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ups, rms := Diff(tc.prev, tc.next)
+			assert.Equal(t, tc.wantUpserts, ups)
+			assert.Equal(t, tc.wantRemoves, rms)
+		})
+	}
+}

internal/agent/inventory/watch.go

@@ -0,0 +1,52 @@
+package inventory
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/docker/docker/api/types/events"
+	"github.com/docker/docker/api/types/filters"
+)
+
+// EventSource is the subset of the Docker SDK Watch needs.
+// *client.Client satisfies it.
+type EventSource interface {
+	Events(ctx context.Context, opts events.ListOptions) (<-chan events.Message, <-chan error)
+}
+
+// Watch subscribes to container and service lifecycle events and calls
+// onChange once after each burst has been quiet for the given duration.
+// A Compose restart emits a dozen events; this turns them into one
+// inventory refresh. Returns nil when ctx ends, or an error if the event
+// stream breaks (for example when the daemon restarts) — the caller should
+// resubscribe and refresh, since events may have been missed.
+func Watch(ctx context.Context, src EventSource, quiet time.Duration, onChange func()) error {
+	args := filters.NewArgs(
+		filters.Arg("type", string(events.ContainerEventType)),
+		filters.Arg("type", string(events.ServiceEventType)),
+	)
+	for _, action := range []string{"create", "start", "stop", "die", "destroy", "rename", "update", "remove"} {
+		args.Add("event", action)
+	}
+	msgs, errs := src.Events(ctx, events.ListOptions{Filters: args})
+
+	timer := time.NewTimer(quiet)
+	timer.Stop()
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case err := <-errs:
+			if ctx.Err() != nil {
+				return nil
+			}
+			return fmt.Errorf("docker events: %w", err)
+		case <-msgs:
+			timer.Reset(quiet)
+		case <-timer.C:
+			onChange()
+		}
+	}
+}

internal/agent/inventory/watch_test.go

@@ -0,0 +1,52 @@
+package inventory
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/docker/docker/api/types/events"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeEvents struct {
+	msgs chan events.Message
+	errs chan error
+}
+
+func newFakeEvents() *fakeEvents {
+	return &fakeEvents{msgs: make(chan events.Message, 10), errs: make(chan error, 1)}
+}
+
+func (f *fakeEvents) Events(context.Context, events.ListOptions) (<-chan events.Message, <-chan error) {
+	return f.msgs, f.errs
+}
+
+func TestWatch_DebouncesBursts(t *testing.T) {
+	src := newFakeEvents()
+	var calls atomic.Int32
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- Watch(ctx, src, 30*time.Millisecond, func() { calls.Add(1) }) }()
+
+	for range 3 {
+		src.msgs <- events.Message{Type: events.ContainerEventType, Action: "start"}
+	}
+	require.Eventually(t, func() bool { return calls.Load() == 1 }, time.Second, 5*time.Millisecond)
+
+	src.msgs <- events.Message{Type: events.ServiceEventType, Action: "update"}
+	require.Eventually(t, func() bool { return calls.Load() == 2 }, time.Second, 5*time.Millisecond)
+
+	cancel()
+	assert.NoError(t, <-done)
+}
+
+func TestWatch_ReturnsStreamError(t *testing.T) {
+	src := newFakeEvents()
+	src.errs <- errors.New("daemon restarted")
+	err := Watch(context.Background(), src, time.Millisecond, func() {})
+	assert.ErrorContains(t, err, "daemon restarted")
+}