Stitch any CI into Tangled
setup eventstream cursors
+19
-5
+19
-5
knot.go
+19
-5
knot.go
···
32
32
33
33
"tangled.org/core/api/tangled"
34
34
"tangled.org/core/eventconsumer"
35
+
"tangled.org/core/eventconsumer/cursor"
35
36
)
36
37
37
38
// KnotConsumer is the surface area the rest of tack uses to interact
···
90
91
// background, and returns the wrapper. The consumer keeps running until
91
92
// ctx is cancelled.
92
93
//
93
-
// Cursor persistence is intentionally in-memory for now: we only log
94
-
// events, so re-receiving a few seconds of pipeline triggers after a
95
-
// restart is harmless. When we start translating triggers into real
96
-
// Buildkite builds, this should switch to a SQLite-backed cursor store
97
-
// to avoid duplicate builds.
94
+
// Cursor persistence is backed by tangled-core's SQLite cursor store
95
+
// pointed at the same database file as the rest of tack. Without
96
+
// this, a restart would replay every recent pipeline event from each
97
+
// knot and fire a duplicate Buildkite build for each one. The
98
+
// cursor.SqliteStore opens its own *sql.DB on the file, but
99
+
// mattn/go-sqlite3 in WAL mode tolerates concurrent connections, and
100
+
// the `cursors` table the upstream package creates doesn't collide
101
+
// with any of our migrations.
98
102
func startKnotConsumer(ctx context.Context, cfg config, st *store, provider Provider) (*knotConsumer, error) {
99
103
logger := loggerFrom(ctx).With("component", "knotconsumer")
100
104
···
103
107
return nil, fmt.Errorf("load known knots: %w", err)
104
108
}
105
109
110
+
// Persistent per-source cursor store. Keyed by source.Key() (the
111
+
// knot hostname for KnotSource), so each knot resumes exactly
112
+
// where it left off and we don't re-fire builds for events that
113
+
// were already processed before the previous shutdown.
114
+
cursorStore, err := cursor.NewSQLiteStore(cfg.DBPath)
115
+
if err != nil {
116
+
return nil, fmt.Errorf("open knot cursor store: %w", err)
117
+
}
118
+
106
119
kc := &knotConsumer{log: logger, provider: provider}
107
120
108
121
ccfg := eventconsumer.NewConsumerConfig()
109
122
ccfg.Logger = logger
110
123
ccfg.Dev = cfg.Dev
111
124
ccfg.ProcessFunc = kc.process
125
+
ccfg.CursorStore = cursorStore
112
126
for _, k := range knots {
113
127
ccfg.Sources[eventconsumer.NewKnotSource(k)] = struct{}{}
114
128
logger.Info("seeding knot source", "knot", k)