diff --git a/platform/metrics/README.md b/platform/metrics/README.md index f3161ac6..25108a34 100644 --- a/platform/metrics/README.md +++ b/platform/metrics/README.md @@ -1,6 +1,6 @@ # Metrics Utilities (`platform/metrics`) -The `metrics` package provides reusable helpers for emitting counters and histograms on a `tally.Scope`. +The `metrics` package provides reusable helpers for emitting counters, gauges, and histograms on a `tally.Scope`. ## Design @@ -48,6 +48,7 @@ For ad-hoc metrics that do not fit the operation lifecycle: | Function | Emits | Example | |----------|-------|---------| | `NamedCounter(scope, name, counter, value, ...tags)` | `{name}.{counter}` counter | `publish.attempts` | +| `NamedGauge(scope, name, gauge, value, ...tags)` | `{name}.{gauge}` gauge | `record.last_green_timestamp_seconds` | | `NamedHistogram(scope, name, histogram, buckets, ...tags)` | `{name}.{histogram}` histogram | `process.duration` | ```go @@ -57,7 +58,9 @@ h := metrics.NamedHistogram(c.scope, "process", "duration", metrics.FastLatencyB h.RecordDuration(elapsed) ``` -Do not emit gauges or timers. Represent operation latency and completion count with lifecycle histograms, and represent instantaneous quantities as sampled histogram values when needed. +Use gauges only for state whose latest value is the whole answer, such as a bookmark timestamp. A gauge is reported once per update rather than continuously, so a gauge set on a discrete event produces a sparse series: it carries no value between updates or after a restart, and each replica reports only the updates it made, so queries must aggregate across replicas with `max` or last-value. State that must be readable at any moment needs a periodic re-emit rather than an event-driven one. + +Represent operation latency and completion count with lifecycle histograms; do not emit timers. ### Why histograms, not timers diff --git a/platform/metrics/metrics.go b/platform/metrics/metrics.go index e2bf0db9..2f506ca5 100644 --- a/platform/metrics/metrics.go +++ b/platform/metrics/metrics.go @@ -171,6 +171,11 @@ func NamedHistogram(scope tally.Scope, name string, histogram string, buckets ta return tagged(scope, tags).SubScope(name).Histogram(histogram, buckets) } +// NamedGauge sets the {name}.{gauge} gauge to value. +func NamedGauge(scope tally.Scope, name string, gauge string, value float64, tags ...Tag) { + tagged(scope, tags).SubScope(name).Gauge(gauge).Update(value) +} + // tagsToMap converts a slice of Tag to a map for tally. func tagsToMap(tags []Tag) map[string]string { m := make(map[string]string, len(tags)) diff --git a/platform/metrics/metrics_test.go b/platform/metrics/metrics_test.go index 4e00bf3c..f6124d9c 100644 --- a/platform/metrics/metrics_test.go +++ b/platform/metrics/metrics_test.go @@ -149,6 +149,15 @@ func TestNamedHistogram(t *testing.T) { assert.True(t, ok, "expected process.duration histogram") } +func TestNamedGauge(t *testing.T) { + scope := tally.NewTestScope("", nil) + NamedGauge(scope, "process", "in_flight", 42, NewTag("queue", "monorepo/main")) + + g, ok := scope.Snapshot().Gauges()["process.in_flight+queue=monorepo/main"] + assert.True(t, ok, "expected tagged process.in_flight gauge") + assert.Equal(t, float64(42), g.Value()) +} + func TestLatencyBuckets_Sorted(t *testing.T) { sets := map[string]tally.DurationBuckets{ "FastLatencyBuckets": FastLatencyBuckets, diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index a3a1f05d..2d3eef57 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -426,7 +426,7 @@ func registerPrimaryControllers( } count++ - recordController := record.NewController(logger, scope, store, stovepipemq.TopicKeyRecord, "stovepipe-record") + recordController := record.NewController(logger, scope, store, scf, stovepipemq.TopicKeyRecord, "stovepipe-record") if err := c.Register(recordController); err != nil { return count, fmt.Errorf("failed to register record controller: %w", err) } diff --git a/stovepipe/controller/record/BUILD.bazel b/stovepipe/controller/record/BUILD.bazel index a66522e3..154efed3 100644 --- a/stovepipe/controller/record/BUILD.bazel +++ b/stovepipe/controller/record/BUILD.bazel @@ -11,6 +11,7 @@ go_library( "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", @@ -26,6 +27,8 @@ go_test( "//platform/consumer/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", + "//stovepipe/extension/sourcecontrol/mock:go_default_library", "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index fe009cab..afca96cb 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -33,6 +33,7 @@ import ( "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) @@ -41,11 +42,12 @@ import ( // advances the queue's last-green bookmark when that fact is green. Implements // consumer.Controller. type Controller struct { - logger *zap.SugaredLogger - metricsScope tally.Scope - stores storage.Factory - topicKey consumer.TopicKey - consumerGroup string + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + sourceControls sourcecontrol.Factory + topicKey consumer.TopicKey + consumerGroup string } // Verify Controller implements consumer.Controller interface at compile time. @@ -64,15 +66,17 @@ func NewController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, + sourceControls sourcecontrol.Factory, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { return &Controller{ - logger: logger.Named("record_controller"), - metricsScope: scope.SubScope("record_controller"), - stores: stores, - topicKey: topicKey, - consumerGroup: consumerGroup, + logger: logger.Named("record_controller"), + metricsScope: scope.SubScope("record_controller"), + stores: stores, + sourceControls: sourceControls, + topicKey: topicKey, + consumerGroup: consumerGroup, } } @@ -256,10 +260,63 @@ func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage "request_id", request.ID, "last_green_uri", request.URI, ) + c.emitLastGreenTimestamp(ctx, request) return nil } } +// emitLastGreenTimestamp emits the creation time of the change the bookmark now +// points at, once that bookmark is durable. Reporting is best-effort so an +// observability failure cannot turn a successful record operation into a retry, +// which is why each cause is counted and logged separately instead of returned. +func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity.Request) { + queueTag := metrics.NewTag("queue", request.Queue) + + sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue}) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_resolve_errors", 1, queueTag) + c.logger.Warnw("failed to resolve source control to report the last green timestamp", + "queue", request.Queue, + "error", err, + ) + return + } + + info, err := sourceControl.ChangeInfo(ctx, request.URI) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_errors", 1, queueTag) + c.logger.Warnw("failed to look up the last green change timestamp", + "queue", request.Queue, + "uri", request.URI, + "error", err, + ) + return + } + + // SourceControl must report a positive creation timestamp, so a missing one + // is a broken extension contract rather than a lookup failure. Emitting it + // anyway would publish a 1970 timestamp and read as an infinitely stale queue. + if info.CreatedAt <= 0 { + metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_invalid", 1, queueTag) + c.logger.Warnw("source control reported no creation timestamp for the last green change", + "queue", request.Queue, + "uri", request.URI, + "created_at", info.CreatedAt, + ) + return + } + + // The gauge carries the creation time as Unix seconds, so subtracting it + // from the current time yields the age of the last-green change in seconds. + metrics.NamedGauge( + c.metricsScope, + _opName, + "last_green_timestamp_seconds", + float64(time.UnixMilli(info.CreatedAt).Unix()), + queueTag, + ) +} + // isNewerRequest reports whether candidate was ingested after current. An empty // current means the bookmark has never been set, so any candidate is newer. func isNewerRequest(queue, candidate, current string) (bool, error) { diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 777b15de..6d0fc5c0 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -18,6 +18,7 @@ import ( "context" "errors" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -26,6 +27,8 @@ import ( consumermock "github.com/uber/submitqueue/platform/consumer/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + sourcecontrolmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock" "github.com/uber/submitqueue/stovepipe/extension/storage" storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" "go.uber.org/mock/gomock" @@ -38,12 +41,16 @@ const ( testURI = "git://remote/monorepo/main/head-sha" ) +var testChangeTime = time.Unix(1_700_000_000, 0).UTC() + // recordMocks bundles the mocks a record controller test case wires // expectations on. type recordMocks struct { - reqStore *storagemock.MockRequestStore - queueStore *storagemock.MockQueueStore - factStore *storagemock.MockValidationFactStore + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore + factStore *storagemock.MockValidationFactStore + sourceControl *sourcecontrolmock.MockSourceControl + metricsScope tally.TestScope } // expectFactCreated wires a successful fact write and captures it, so a case can @@ -62,13 +69,31 @@ type staticStorageFactory struct{ store storage.Storage } // For returns the fixed store aggregate for any queue. func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil } +type staticSourceControlFactory struct { + sourceControl sourcecontrol.SourceControl +} + +func (f staticSourceControlFactory) For(sourcecontrol.Config) (sourcecontrol.SourceControl, error) { + return f.sourceControl, nil +} + +// failingSourceControlFactory resolves no queue. +type failingSourceControlFactory struct{} + +func (failingSourceControlFactory) For(sourcecontrol.Config) (sourcecontrol.SourceControl, error) { + return nil, errors.New("no source control for queue") +} + func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMocks) { t.Helper() + scope := tally.NewTestScope("", nil) m := recordMocks{ - reqStore: storagemock.NewMockRequestStore(ctrl), - queueStore: storagemock.NewMockQueueStore(ctrl), - factStore: storagemock.NewMockValidationFactStore(ctrl), + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + factStore: storagemock.NewMockValidationFactStore(ctrl), + sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), + metricsScope: scope, } store := storagemock.NewMockStorage(ctrl) @@ -78,8 +103,9 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMo c := NewController( zap.NewNop().Sugar(), - tally.NewTestScope("test", nil), + scope, staticStorageFactory{store: store}, + staticSourceControlFactory{sourceControl: m.sourceControl}, stovepipemq.TopicKeyRecord, "stovepipe-record", ) @@ -158,11 +184,17 @@ func TestProcess_AdvancesBookmarkOnSuccess(t *testing.T) { written = q return nil }) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) assert.Equal(t, tt.wantURI, written.LastGreenURI) assert.Equal(t, testID, written.LastGreenRequestID) + gauge, ok := m.metricsScope.Snapshot().Gauges()["record_controller.record.last_green_timestamp_seconds+queue=monorepo/main"] + require.True(t, ok) + assert.Equal(t, float64(testChangeTime.Unix()), gauge.Value()) + // The green fact is what authorises the advance. assert.Equal(t, entity.DegreeGreen, fact.Degree) assert.Equal(t, testURI, fact.URI) @@ -173,6 +205,68 @@ func TestProcess_AdvancesBookmarkOnSuccess(t *testing.T) { } } +func TestProcess_TimestampReportingFailureDoesNotFailRecord(t *testing.T) { + tests := []struct { + name string + info sourcecontrol.ChangeInfo + err error + wantCounter string + }{ + { + name: "lookup fails", + err: errors.New("boom"), + wantCounter: "record_controller.record.last_green_timestamp_errors+queue=monorepo/main", + }, + { + // A zero timestamp breaks the extension contract, so it is counted + // apart from a lookup failure rather than emitted as a 1970 gauge. + name: "timestamp missing", + info: sourcecontrol.ChangeInfo{CreatedAt: 0}, + wantCounter: "record_controller.record.last_green_timestamp_invalid+queue=monorepo/main", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil) + m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI).Return(tt.info, tt.err) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) + assert.Empty(t, m.metricsScope.Snapshot().Gauges()) + counter, ok := m.metricsScope.Snapshot().Counters()[tt.wantCounter] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) + }) + } +} + +func TestProcess_UnresolvableSourceControlDoesNotFailRecord(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + c.sourceControls = failingSourceControlFactory{} + + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil) + m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) + assert.Empty(t, m.metricsScope.Snapshot().Gauges()) + counter, ok := m.metricsScope.Snapshot().Counters()["record_controller.record.last_green_timestamp_resolve_errors+queue=monorepo/main"] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) +} + func TestProcess_RecordsBrokenFactWithoutAdvancing(t *testing.T) { ctrl := gomock.NewController(t) c, m := newController(t, ctrl) @@ -221,6 +315,8 @@ func TestProcess_AdoptsExistingFactFromSameRequest(t *testing.T) { if tt.wantUpdate { m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil) m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) } require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) @@ -270,6 +366,11 @@ func TestProcess_SkipsBookmarkWhenNotNewer(t *testing.T) { // No Update: the bookmark only moves forward. require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) + assert.NotContains( + t, + m.metricsScope.Snapshot().Gauges(), + "record_controller.record.last_green_timestamp_seconds+queue=monorepo/main", + ) }) } } @@ -338,6 +439,8 @@ func TestProcess_RetriesBookmarkOnVersionMismatch(t *testing.T) { m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), fresh.Version, fresh.Version+1). Return(nil), ) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI). + Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil) require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) } diff --git a/stovepipe/extension/sourcecontrol/fake/fake.go b/stovepipe/extension/sourcecontrol/fake/fake.go index ae57a8c1..5890a906 100644 --- a/stovepipe/extension/sourcecontrol/fake/fake.go +++ b/stovepipe/extension/sourcecontrol/fake/fake.go @@ -21,17 +21,24 @@ package fake import ( "context" + "time" "github.com/uber/submitqueue/platform/base/page" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" ) +// changeInterval is the synthetic spacing between adjacent history entries. +const changeInterval = time.Minute + // sourceControlFake serves a single queue's linear history. history[0] is the // latest commit; higher indices are progressively older ancestors. type sourceControlFake struct { // cfg is the per-queue identity this source control was built for. cfg sourcecontrol.Config history []string + // latestCreatedAt is the millisecond timestamp assigned to history[0]; older + // ancestors are spaced changeInterval apart behind it. + latestCreatedAt int64 } // New returns a sourcecontrol.SourceControl bound to the queue named in cfg, @@ -41,7 +48,11 @@ type sourceControlFake struct { func New(cfg sourcecontrol.Config, history []string) sourcecontrol.SourceControl { cp := make([]string, len(history)) copy(cp, history) - return sourceControlFake{cfg: cfg, history: cp} + return sourceControlFake{ + cfg: cfg, + history: cp, + latestCreatedAt: time.Now().UnixMilli(), + } } // Latest returns the newest commit URI, or ErrNotFound when the history is empty. @@ -94,6 +105,17 @@ func (s sourceControlFake) History(_ context.Context, cursor string, limit int) return page.Page[string]{Items: uris, NextCursor: next}, nil } +// ChangeInfo returns immutable metadata for a change on the fake's ref. +func (s sourceControlFake) ChangeInfo(_ context.Context, uri string) (sourcecontrol.ChangeInfo, error) { + index := s.indexOf(uri) + if index < 0 { + return sourcecontrol.ChangeInfo{}, sourcecontrol.ErrNotFound + } + return sourcecontrol.ChangeInfo{ + CreatedAt: s.latestCreatedAt - int64(index)*changeInterval.Milliseconds(), + }, nil +} + // indexOf returns the index of uri in the history, or -1 if absent. func (s sourceControlFake) indexOf(uri string) int { for i, u := range s.history { diff --git a/stovepipe/extension/sourcecontrol/fake/fake_test.go b/stovepipe/extension/sourcecontrol/fake/fake_test.go index 77b3bab3..2e2af523 100644 --- a/stovepipe/extension/sourcecontrol/fake/fake_test.go +++ b/stovepipe/extension/sourcecontrol/fake/fake_test.go @@ -17,6 +17,7 @@ package fake import ( "context" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -140,3 +141,18 @@ func TestHistory(t *testing.T) { }) } } + +func TestChangeInfo(t *testing.T) { + source := New(testCfg, history) + + latest, err := source.ChangeInfo(context.Background(), "git://repo/ref/c") + require.NoError(t, err) + assert.InDelta(t, time.Now().UnixMilli(), latest.CreatedAt, float64(time.Minute.Milliseconds())) + + oldest, err := source.ChangeInfo(context.Background(), "git://repo/ref/a") + require.NoError(t, err) + assert.Equal(t, 2*changeInterval.Milliseconds(), latest.CreatedAt-oldest.CreatedAt) + + _, err = source.ChangeInfo(context.Background(), "git://repo/ref/x") + require.ErrorIs(t, err, sourcecontrol.ErrNotFound) +} diff --git a/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go b/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go index 92d52526..a7a3c05f 100644 --- a/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go +++ b/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go @@ -42,6 +42,21 @@ func (m *MockSourceControl) EXPECT() *MockSourceControlMockRecorder { return m.recorder } +// ChangeInfo mocks base method. +func (m *MockSourceControl) ChangeInfo(ctx context.Context, uri string) (sourcecontrol.ChangeInfo, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ChangeInfo", ctx, uri) + ret0, _ := ret[0].(sourcecontrol.ChangeInfo) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ChangeInfo indicates an expected call of ChangeInfo. +func (mr *MockSourceControlMockRecorder) ChangeInfo(ctx, uri any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChangeInfo", reflect.TypeOf((*MockSourceControl)(nil).ChangeInfo), ctx, uri) +} + // History mocks base method. func (m *MockSourceControl) History(ctx context.Context, cursor string, limit int) (page.Page[string], error) { m.ctrl.T.Helper() diff --git a/stovepipe/extension/sourcecontrol/sourcecontrol.go b/stovepipe/extension/sourcecontrol/sourcecontrol.go index d61a1013..3dccec03 100644 --- a/stovepipe/extension/sourcecontrol/sourcecontrol.go +++ b/stovepipe/extension/sourcecontrol/sourcecontrol.go @@ -32,6 +32,15 @@ import ( "github.com/uber/submitqueue/platform/base/page" ) +// ChangeInfo describes immutable metadata about the source-control change +// represented by a URI. +type ChangeInfo struct { + // CreatedAt is the millisecond timestamp at which the source-control + // provider recorded the immutable change. It does not represent ref-update + // or fetch time. Always positive for a resolvable URI. + CreatedAt int64 +} + // ErrNotFound is returned when a queue, ref, or URI cannot be resolved by the // implementation (for example an unknown queue, or an ancestry query referencing // a URI that is not on the ref). @@ -71,6 +80,10 @@ type SourceControl interface { // the greenness/status of each commit. Returns ErrNotFound if the cursor does // not refer to a position on the ref. History(ctx context.Context, cursor string, limit int) (page.Page[string], error) + + // ChangeInfo returns immutable metadata for uri. The returned CreatedAt must + // be positive. Returns ErrNotFound when uri cannot be resolved. + ChangeInfo(ctx context.Context, uri string) (ChangeInfo, error) } // Config carries the per-queue identity handed to a Factory. The system knows