Stitch any CI into Tangled
seed initial event consumer cursor to now to avoid historical records
+46
-1
+46
-1
knot.go
+46
-1
knot.go
···
29
29
"encoding/json"
30
30
"fmt"
31
31
"log/slog"
32
+
"time"
32
33
33
34
"tangled.org/core/api/tangled"
34
35
"tangled.org/core/eventconsumer"
···
75
76
c *eventconsumer.Consumer
76
77
log *slog.Logger
77
78
79
+
// cursors is the persistent cursor store the underlying consumer
80
+
// reads at every (re)connect. We retain a reference here so we
81
+
// can pre-seed an entry to "now" the first time we ever see a
82
+
// knot — see seedCursorIfMissing for why.
83
+
cursors cursor.Store
84
+
78
85
// provider dispatches each incoming pipeline trigger to whatever
79
86
// backend actually runs it (today: the fake provider; tomorrow:
80
87
// Buildkite). The consumer doesn't care which — it just hands
···
116
123
return nil, fmt.Errorf("open knot cursor store: %w", err)
117
124
}
118
125
119
-
kc := &knotConsumer{log: logger, provider: provider}
126
+
kc := &knotConsumer{log: logger, cursors: cursorStore, provider: provider}
120
127
121
128
ccfg := eventconsumer.NewConsumerConfig()
122
129
ccfg.Logger = logger
···
124
131
ccfg.ProcessFunc = kc.process
125
132
ccfg.CursorStore = cursorStore
126
133
for _, k := range knots {
134
+
// Pin a brand-new knot's cursor to "now" before the consumer
135
+
// ever connects, so a fresh tack install doesn't replay every
136
+
// historical pipeline event the knot has retained and fire a
137
+
// duplicate Buildkite build for each one.
138
+
kc.seedCursorIfMissing(k)
127
139
ccfg.Sources[eventconsumer.NewKnotSource(k)] = struct{}{}
128
140
logger.Info("seeding knot source", "knot", k)
129
141
}
···
145
157
if knot == "" {
146
158
return
147
159
}
160
+
// Same first-run protection as in startKnotConsumer: a knot we
161
+
// have never observed before must not retroactively fire
162
+
// pipelines for triggers older than the moment we learned about
163
+
// it. AddSource reads the cursor synchronously when it dials.
164
+
k.seedCursorIfMissing(knot)
148
165
k.log.Info("adding knot source", "knot", knot)
149
166
k.c.AddSource(ctx, eventconsumer.NewKnotSource(knot))
167
+
}
168
+
169
+
// seedCursorIfMissing writes the current time (as nanoseconds since
170
+
// the unix epoch, the same unit eventconsumer.worker uses when it
171
+
// advances cursors after each message) to the cursor store for knot
172
+
// iff no cursor is already persisted for it.
173
+
//
174
+
// Without this, a brand-new install — or the first time we ever
175
+
// dial a previously-unseen knot — connects to /events with no
176
+
// cursor query parameter, which the knot servers interpret as
177
+
// "stream from the beginning of time." For pipeline records that
178
+
// would fire one Buildkite build per historical trigger the knot
179
+
// still has retained. Pinning the cursor up-front limits us to
180
+
// events that arrive *after* tack learned about the knot.
181
+
//
182
+
// Get returning 0 means "no cursor stored" across all the cursor
183
+
// store implementations we use (memory/sqlite/redis), so it's a
184
+
// safe sentinel to gate the write.
185
+
func (k *knotConsumer) seedCursorIfMissing(knot string) {
186
+
if k.cursors.Get(knot) != 0 {
187
+
return
188
+
}
189
+
now := time.Now().UnixNano()
190
+
k.cursors.Set(knot, now)
191
+
k.log.Info("seeded fresh knot cursor to now",
192
+
"knot", knot,
193
+
"cursor", now,
194
+
)
150
195
}
151
196
152
197
// RemoveKnot is currently a no-op — see the interface comment for the