remove knot reconciliation · jcollie.dev/tack@72776b7

+80 -12

jetstream.go

··· 215 215 if err := json.Unmarshal(c.Record, &rec); err != nil { 216 216 return fmt.Errorf("decode repo: %w", err) 217 217 } 218 + 219 + // Capture the prior (knot, spindle) before the upsert so the 220 + // post-mutation reconcile below can detect transitions like 221 + // "repo used to point at us, no longer does" — which would 222 + // otherwise leave a knot subscription dangling. 223 + oldKnot, oldSpindle, err := st.GetRepo(ctx, did, c.RKey) 224 + if err != nil { 225 + return err 226 + } 227 + 218 228 if err := st.UpsertRepo(ctx, did, c.RKey, 219 229 rec.Knot, rec.Name, 220 230 deref(rec.Spindle), deref(rec.RepoDid), ··· 223 233 return err 224 234 } 225 235 226 - // If this repo just declared us as its spindle, start (or 227 - // continue) listening to its knot for pipeline triggers. The 228 - // knot consumer dedupes on its own so this is safe to call 229 - // even on update events that don't change the spindle field. 230 - if knots != nil && rec.Spindle != nil && *rec.Spindle == hostname && rec.Knot != "" { 231 - knots.AddKnot(ctx, rec.Knot) 236 + newSpindle := deref(rec.Spindle) 237 + return reconcileKnot(ctx, st, knots, hostname, 238 + oldKnot, oldSpindle, 239 + rec.Knot, newSpindle, 240 + ) 241 + 242 + case jsOpDelete: 243 + // Same shape as the update path, just with no "new" side: we 244 + // have to read the row out before deleting so we can decide 245 + // whether deletion freed the last hold on a knot we'd been 246 + // subscribed to. 247 + oldKnot, oldSpindle, err := st.GetRepo(ctx, did, c.RKey) 248 + if err != nil { 249 + return err 232 250 } 251 + if err := st.DeleteRepo(ctx, did, c.RKey); err != nil { 252 + return err 253 + } 254 + return reconcileKnot(ctx, st, knots, hostname, 255 + oldKnot, oldSpindle, 256 + "", "", 257 + ) 258 + } 259 + return nil 260 + } 233 261 262 + // reconcileKnot brings the knot consumer's subscriptions in line with 263 + // the latest store state after a single repo mutation. It is called 264 + // after the mutation has been applied so IsKnotWanted reflects the 265 + // post-mutation truth. 266 + // 267 + // Logic: 268 + // - If the new record names us as its spindle, ensure we're 269 + // subscribed to its knot (AddKnot is idempotent, so calling it on 270 + // an already-watched knot is cheap). 271 + // - If the old record named us as its spindle, check whether any 272 + // other repo still references that knot through us; if not, 273 + // RemoveKnot it. Skip this when the knot didn't change AND the 274 + // spindle didn't move away from us, because then nothing actually 275 + // released our hold. 276 + func reconcileKnot( 277 + ctx context.Context, 278 + st *store, 279 + knots KnotConsumer, 280 + hostname string, 281 + oldKnot, oldSpindle string, 282 + newKnot, newSpindle string, 283 + ) error { 284 + // Tests pass nil for the consumer when they only care about the 285 + // store mutation half of the handler; tolerate that here so 286 + // callers don't have to special-case it. 287 + if knots == nil { 234 288 return nil 235 - case jsOpDelete: 236 - // We don't unsubscribe from the knot here: other repos may 237 - // still want us to watch it. A periodic reconciliation pass 238 - // (not yet implemented) is the right place to drop unused 239 - // subscriptions. 240 - return st.DeleteRepo(ctx, did, c.RKey) 289 + } 290 + 291 + if newSpindle == hostname && newKnot != "" { 292 + knots.AddKnot(ctx, newKnot) 293 + } 294 + 295 + // Did we just lose our claim on oldKnot? Two ways that can happen: 296 + // the spindle field moved off of us, or the knot field moved to a 297 + // different host. Either is a reason to consider unsubscribing 298 + // from oldKnot — but only if no *other* repo still has us on it. 299 + releasedOld := oldSpindle == hostname && oldKnot != "" && 300 + (newSpindle != hostname || newKnot != oldKnot) 301 + if releasedOld { 302 + stillWanted, err := st.IsKnotWanted(ctx, hostname, oldKnot) 303 + if err != nil { 304 + return err 305 + } 306 + if !stillWanted { 307 + knots.RemoveKnot(ctx, oldKnot) 308 + } 241 309 } 242 310 return nil 243 311 }

+161

jetstream_test.go

··· 310 310 t.Fatalf("AddKnot calls = %v, want none", added) 311 311 } 312 312 } 313 + 314 + // TestRepoUpdateSpindleAwayFromUsRemovesKnot covers the case where a 315 + // repo we'd previously been watching gets its .spindle field flipped to 316 + // some other spindle. Once that's the only repo we had on that knot, 317 + // the reconciliation must call RemoveKnot. 318 + func TestRepoUpdateSpindleAwayFromUsRemovesKnot(t *testing.T) { 319 + s := newTestStore(t) 320 + ctx := context.Background() 321 + 322 + const ours = "tack.example" 323 + const knot = "knot.example" 324 + 325 + // Seed: a repo that names us as its spindle on `knot`. 326 + if err := s.UpsertRepo(ctx, "did:plc:a", "rk", knot, "repo-a", ours, "", "t"); err != nil { 327 + t.Fatal(err) 328 + } 329 + 330 + // Update: same record, now points at a different spindle. 331 + other := "other.example" 332 + rec := tangled.Repo{ 333 + Knot: knot, 334 + Name: "repo-a", 335 + Spindle: &other, 336 + CreatedAt: "2026-01-01T00:00:00Z", 337 + } 338 + evt := commitEvent(1, "did:plc:a", tangled.RepoNSID, jsOpUpdate, "rk", rec) 339 + 340 + fake := &fakeKnotConsumer{} 341 + if err := handleJetstreamEvent(ctx, s, fake, ours, evt); err != nil { 342 + t.Fatalf("handle: %v", err) 343 + } 344 + if added := fake.Added(); len(added) != 0 { 345 + t.Fatalf("AddKnot calls = %v, want none", added) 346 + } 347 + if removed := fake.Removed(); len(removed) != 1 || removed[0] != knot { 348 + t.Fatalf("RemoveKnot calls = %v, want [%s]", removed, knot) 349 + } 350 + } 351 + 352 + // TestRepoUpdateSpindleAwayFromUsKeepsKnotIfShared ensures we don't 353 + // over-eagerly unsubscribe from a knot when one of multiple repos on 354 + // that knot moves away from us. The other repo's subscription must keep 355 + // the knot in our wanted set. 356 + func TestRepoUpdateSpindleAwayFromUsKeepsKnotIfShared(t *testing.T) { 357 + s := newTestStore(t) 358 + ctx := context.Background() 359 + 360 + const ours = "tack.example" 361 + const knot = "knot.example" 362 + 363 + // Two repos sharing one knot, both pointed at us. 364 + if err := s.UpsertRepo(ctx, "did:plc:a", "rk1", knot, "a", ours, "", "t"); err != nil { 365 + t.Fatal(err) 366 + } 367 + if err := s.UpsertRepo(ctx, "did:plc:b", "rk2", knot, "b", ours, "", "t"); err != nil { 368 + t.Fatal(err) 369 + } 370 + 371 + // Repo A flips to a different spindle. B still wants us. 372 + other := "other.example" 373 + rec := tangled.Repo{ 374 + Knot: knot, 375 + Name: "a", 376 + Spindle: &other, 377 + CreatedAt: "2026-01-01T00:00:00Z", 378 + } 379 + evt := commitEvent(1, "did:plc:a", tangled.RepoNSID, jsOpUpdate, "rk1", rec) 380 + 381 + fake := &fakeKnotConsumer{} 382 + if err := handleJetstreamEvent(ctx, s, fake, ours, evt); err != nil { 383 + t.Fatalf("handle: %v", err) 384 + } 385 + if removed := fake.Removed(); len(removed) != 0 { 386 + t.Fatalf("RemoveKnot calls = %v, want none (B still wants us on %s)", removed, knot) 387 + } 388 + } 389 + 390 + // TestRepoUpdateChangingKnotSwapsSubscription verifies that a repo 391 + // staying with us but changing its .knot field unsubscribes the old 392 + // knot (if no other repo holds it) and subscribes the new one. 393 + func TestRepoUpdateChangingKnotSwapsSubscription(t *testing.T) { 394 + s := newTestStore(t) 395 + ctx := context.Background() 396 + 397 + const ours = "tack.example" 398 + const oldKnot = "old.example" 399 + const newKnot = "new.example" 400 + 401 + if err := s.UpsertRepo(ctx, "did:plc:a", "rk", oldKnot, "a", ours, "", "t"); err != nil { 402 + t.Fatal(err) 403 + } 404 + 405 + spindle := ours 406 + rec := tangled.Repo{ 407 + Knot: newKnot, 408 + Name: "a", 409 + Spindle: &spindle, 410 + CreatedAt: "2026-01-01T00:00:00Z", 411 + } 412 + evt := commitEvent(1, "did:plc:a", tangled.RepoNSID, jsOpUpdate, "rk", rec) 413 + 414 + fake := &fakeKnotConsumer{} 415 + if err := handleJetstreamEvent(ctx, s, fake, ours, evt); err != nil { 416 + t.Fatalf("handle: %v", err) 417 + } 418 + if added := fake.Added(); len(added) != 1 || added[0] != newKnot { 419 + t.Fatalf("AddKnot calls = %v, want [%s]", added, newKnot) 420 + } 421 + if removed := fake.Removed(); len(removed) != 1 || removed[0] != oldKnot { 422 + t.Fatalf("RemoveKnot calls = %v, want [%s]", removed, oldKnot) 423 + } 424 + } 425 + 426 + // TestRepoDeleteRemovesKnotWhenLast confirms deleting the last repo on 427 + // a knot we cared about triggers RemoveKnot. 428 + func TestRepoDeleteRemovesKnotWhenLast(t *testing.T) { 429 + s := newTestStore(t) 430 + ctx := context.Background() 431 + 432 + const ours = "tack.example" 433 + const knot = "knot.example" 434 + 435 + if err := s.UpsertRepo(ctx, "did:plc:a", "rk", knot, "a", ours, "", "t"); err != nil { 436 + t.Fatal(err) 437 + } 438 + evt := commitEvent(1, "did:plc:a", tangled.RepoNSID, jsOpDelete, "rk", nil) 439 + 440 + fake := &fakeKnotConsumer{} 441 + if err := handleJetstreamEvent(ctx, s, fake, ours, evt); err != nil { 442 + t.Fatalf("handle: %v", err) 443 + } 444 + if removed := fake.Removed(); len(removed) != 1 || removed[0] != knot { 445 + t.Fatalf("RemoveKnot calls = %v, want [%s]", removed, knot) 446 + } 447 + } 448 + 449 + // TestRepoDeleteKeepsKnotIfShared ensures deleting one of multiple 450 + // repos on a knot does not unsubscribe — the survivors still want it. 451 + func TestRepoDeleteKeepsKnotIfShared(t *testing.T) { 452 + s := newTestStore(t) 453 + ctx := context.Background() 454 + 455 + const ours = "tack.example" 456 + const knot = "knot.example" 457 + 458 + if err := s.UpsertRepo(ctx, "did:plc:a", "rk1", knot, "a", ours, "", "t"); err != nil { 459 + t.Fatal(err) 460 + } 461 + if err := s.UpsertRepo(ctx, "did:plc:b", "rk2", knot, "b", ours, "", "t"); err != nil { 462 + t.Fatal(err) 463 + } 464 + evt := commitEvent(1, "did:plc:a", tangled.RepoNSID, jsOpDelete, "rk1", nil) 465 + 466 + fake := &fakeKnotConsumer{} 467 + if err := handleJetstreamEvent(ctx, s, fake, ours, evt); err != nil { 468 + t.Fatalf("handle: %v", err) 469 + } 470 + if removed := fake.Removed(); len(removed) != 0 { 471 + t.Fatalf("RemoveKnot calls = %v, want none", removed) 472 + } 473 + }

+27 -4

knot.go

··· 19 19 // Once the build pipeline is wired up this is where pipeline 20 20 // triggers will be translated into Buildkite builds. 21 21 // 22 - // The jetstream consumer also gets a back-reference (via the knotAdder 23 - // interface) so it can dynamically subscribe to a new knot the moment a 24 - // matching sh.tangled.repo record arrives, without waiting for a tack 25 - // restart. 22 + // The jetstream consumer also gets a back-reference (via the 23 + // KnotConsumer interface) so it can dynamically subscribe a new knot — 24 + // or, conceptually, unsubscribe one — the moment a matching 25 + // sh.tangled.repo record arrives, without waiting for a tack restart. 26 26 27 27 import ( 28 28 "context" ··· 50 50 // a no-op. An empty knot string is ignored. The supplied context 51 51 // scopes the dial; cancelling it tears the subscription down. 52 52 AddKnot(ctx context.Context, knot string) 53 + 54 + // RemoveKnot stops processing events from the given knot. It is the 55 + // inverse of AddKnot and must tolerate being called for a knot that 56 + // was never added (no-op). An empty knot string is ignored. 57 + // 58 + // The production implementation is currently a no-op: tangled-core's 59 + // eventconsumer does not expose a way to drop an individual source's 60 + // websocket. Tracked upstream as 61 + // https://tangled.org/did:plc:j5hmlfdrwkvtxm7cjmu7j2is/issues/510 62 + RemoveKnot(ctx context.Context, knot string) 53 63 } 54 64 55 65 // knotConsumer is the production KnotConsumer. It wraps ··· 116 126 } 117 127 k.log.Info("adding knot source", "knot", knot) 118 128 k.c.AddSource(ctx, eventconsumer.NewKnotSource(knot)) 129 + } 130 + 131 + // RemoveKnot is currently a no-op — see the interface comment for the 132 + // upstream blocker. We still log at info so operators can see when the 133 + // reconciliation logic *would* have unsubscribed; once eventconsumer 134 + // gains a RemoveSource primitive, swap the body for a real call. 135 + func (k *knotConsumer) RemoveKnot(_ context.Context, knot string) { 136 + if knot == "" { 137 + return 138 + } 139 + k.log.Info("remove knot source requested (no-op until upstream supports it)", 140 + "knot", knot, 141 + ) 119 142 } 120 143 121 144 // Stop tears down all knot websocket connections and waits for the

+23 -5

knot_fake.go

··· 6 6 // any future test files (and, if we ever split tack into subpackages, can 7 7 // be promoted to an exported helper without moving code around). 8 8 // 9 - // It does no I/O: AddKnot just records the knot it was handed so tests 10 - // can assert on the side effect. 9 + // It does no I/O: AddKnot and RemoveKnot just record the knot they were 10 + // handed so tests can assert on the side effect. 11 11 12 12 import ( 13 13 "context" ··· 15 15 ) 16 16 17 17 // fakeKnotConsumer is an in-memory KnotConsumer suitable for tests. The 18 - // zero value is ready to use; concurrent calls to AddKnot are safe. 18 + // zero value is ready to use; concurrent calls are safe. 19 19 type fakeKnotConsumer struct { 20 - mu sync.Mutex 21 - added []string 20 + mu sync.Mutex 21 + added []string 22 + removed []string 22 23 } 23 24 24 25 // Compile-time interface conformance check — keeps the fake honest if ··· 32 33 f.added = append(f.added, knot) 33 34 } 34 35 36 + // RemoveKnot records the knot for later inspection via Removed(). 37 + func (f *fakeKnotConsumer) RemoveKnot(_ context.Context, knot string) { 38 + f.mu.Lock() 39 + defer f.mu.Unlock() 40 + f.removed = append(f.removed, knot) 41 + } 42 + 35 43 // Added returns a copy of the knots passed to AddKnot, in call order. 36 44 // A copy is returned so callers can't accidentally mutate the fake's 37 45 // internal slice while comparing. ··· 42 50 copy(out, f.added) 43 51 return out 44 52 } 53 + 54 + // Removed returns a copy of the knots passed to RemoveKnot, in call 55 + // order. See Added() for the rationale behind copying. 56 + func (f *fakeKnotConsumer) Removed() []string { 57 + f.mu.Lock() 58 + defer f.mu.Unlock() 59 + out := make([]string, len(f.removed)) 60 + copy(out, f.removed) 61 + return out 62 + }

+39

store.go

··· 194 194 return nil 195 195 } 196 196 197 + // GetRepo returns the (knot, spindle) currently stored for a (did, rkey) 198 + // pair. Both are returned as empty strings when no row exists; callers 199 + // that need to distinguish "absent" from "stored but empty" should 200 + // pre-check existence themselves. 201 + // 202 + // This exists so applyRepo can read the *previous* spindle/knot of a 203 + // record before applying a mutation, which is what makes it possible to 204 + // detect transitions like "this repo used to be ours, now it isn't" and 205 + // trigger a knot unsubscribe. 206 + func (s *store) GetRepo(ctx context.Context, did, rkey string) (knot, spindle string, err error) { 207 + err = s.db.QueryRowContext(ctx, 208 + `SELECT knot, spindle FROM repos WHERE did = ? AND rkey = ?`, 209 + did, rkey, 210 + ).Scan(&knot, &spindle) 211 + if errors.Is(err, sql.ErrNoRows) { 212 + return "", "", nil 213 + } 214 + if err != nil { 215 + return "", "", fmt.Errorf("get repo: %w", err) 216 + } 217 + return knot, spindle, nil 218 + } 219 + 220 + // IsKnotWanted reports whether any repo currently stored still names the 221 + // given hostname as its spindle and the given knot as its host. After a 222 + // repo update or delete this is the question we ask to decide whether 223 + // to keep watching that knot or unsubscribe from it. 224 + func (s *store) IsKnotWanted(ctx context.Context, hostname, knot string) (bool, error) { 225 + var n int 226 + err := s.db.QueryRowContext(ctx, 227 + `SELECT COUNT(*) FROM repos WHERE spindle = ? AND knot = ?`, 228 + hostname, knot, 229 + ).Scan(&n) 230 + if err != nil { 231 + return false, fmt.Errorf("count repos for knot: %w", err) 232 + } 233 + return n > 0, nil 234 + } 235 + 197 236 // KnotsForSpindle returns the distinct knot hostnames of all repos that 198 237 // have declared the given spindle hostname as their CI spindle. The knot 199 238 // event-stream subscriber uses this to decide which knots to dial.

Configure Feed

Configure Feed