From e34a5cbbc2bc4ec5ab0a8311ba7785f5d0f2f899 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 4 Oct 2026 09:02:33 -0400 Subject: [PATCH 1/3] fix(runnerhub): archive enroll-fallback sessions under their tenant (RIG-4296) When the durable sweep faults, enroll archives the RAM snapshot of the Runner's sessions. It ran those archives under the enroll ctx, which has no tenant. Each archive now resolves the session's tenant through runnerSessionCtx; the failed sweep left the rows, so the lookup still works. Main already reaps every tenant under the system role (RIG-2872), so this change covers only the fallback path. A pgtest pins each swept session archiving under its own tenant, the bootstrap tenant included. Co-authored-by: Matt Wilkinson --- go/internal/runnerhub/hub.go | 5 +- .../runnerhub/runner_tenant_pgtest_test.go | 87 +++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index fc2ada08b..6aa434548 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -1073,8 +1073,11 @@ func (h *Hub) enroll(ctx context.Context, id string, subject store.Subject, tier h.archiveEnded(store.WithTenant(ctx, b.TenantID), b.SessionID) } } else { + // The failed sweep left the rows, so each session's tenant is still readable. for _, sessionID := range reapedSessions { - h.archiveEnded(ctx, sessionID) + if sctx, ok := h.runnerSessionCtx(ctx, id, sessionID); ok { + h.archiveEnded(sctx, sessionID) + } } } return reattached diff --git a/go/internal/runnerhub/runner_tenant_pgtest_test.go b/go/internal/runnerhub/runner_tenant_pgtest_test.go index e326caa65..894cfaf3b 100644 --- a/go/internal/runnerhub/runner_tenant_pgtest_test.go +++ b/go/internal/runnerhub/runner_tenant_pgtest_test.go @@ -4,6 +4,8 @@ package runnerhub import ( "context" + "errors" + "maps" "testing" "time" @@ -14,6 +16,91 @@ import ( "github.com/RigelBuild/compass/go/internal/store" ) +type tenantRecordingEndSink struct { + ended chan struct { + tenant store.TenantID + sessionID string + } +} + +func (s tenantRecordingEndSink) OnSessionEnded(ctx context.Context, sessionID string) { + tenant, _ := store.TenantFromContext(ctx) + s.ended <- struct { + tenant store.TenantID + sessionID string + }{tenant: tenant, sessionID: sessionID} +} + +func TestEnrollReapsEveryTenantsBindings(t *testing.T) { + ctx := context.Background() + dsn := pgtest.RequireDSN(t) + st, err := store.Open(ctx, dsn) + if err != nil { + t.Fatalf("store.Open: %v", err) + } + t.Cleanup(st.Close) + bootstrap, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: "admin", DisplayName: "admin"}) + if err != nil { + t.Fatalf("BootstrapAdmin: %v", err) + } + bootstrapAgent, err := st.CreateAgent(ctx, bootstrap.ID, store.NewAgent{Handle: "agent-bootstrap", DisplayName: "agent-bootstrap"}) + if err != nil { + t.Fatalf("CreateAgent bootstrap: %v", err) + } + tenantB := seedTenantRow(t, ctx, dsn, "tenant-b-reap") + ctxB := store.WithTenant(ctx, tenantB) + owner, err := st.CreateUser(ctxB, store.NewUser{Handle: "owner-b-reap", DisplayName: "owner-b-reap"}) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + agent, err := st.CreateAgent(ctxB, owner.ID, store.NewAgent{Handle: "agent-b-reap", DisplayName: "agent-b-reap"}) + if err != nil { + t.Fatalf("CreateAgent: %v", err) + } + if _, err := st.RecordSessionBinding(ctxB, "sess-b", agent.ID, "runner-1"); err != nil { + t.Fatalf("RecordSessionBinding tenant B: %v", err) + } + if _, err := st.RecordSessionBinding(ctx, "sess-bootstrap", bootstrapAgent.ID, "runner-1"); err != nil { + t.Fatalf("RecordSessionBinding bootstrap: %v", err) + } + + hub := newHubOnly() + hub.SetSessionBindingStore(st) + ended := tenantRecordingEndSink{ended: make(chan struct { + tenant store.TenantID + sessionID string + }, 2)} + hub.SetSessionEndSink(ended) + hub.enroll(ctx, "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) + + if _, _, err := st.ResolveSessionBinding(ctxB, "sess-b"); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("tenant B binding resolve error = %v, want ErrNotFound", err) + } + if _, _, err := st.ResolveSessionBinding(ctx, "sess-bootstrap"); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("bootstrap binding resolve error = %v, want ErrNotFound", err) + } + bootstrapTenant, err := st.BootstrapTenant(ctx) + if err != nil { + t.Fatalf("BootstrapTenant: %v", err) + } + want := map[string]store.TenantID{"sess-b": tenantB, "sess-bootstrap": bootstrapTenant} + got := make(map[string]store.TenantID, len(want)) + for range len(want) { + select { + case e := <-ended.ended: + if _, dup := got[e.sessionID]; dup { + t.Fatalf("session %q archived twice", e.sessionID) + } + got[e.sessionID] = e.tenant + case <-time.After(10 * time.Second): + t.Fatalf("session-end archives = %v, want %v", got, want) + } + } + if !maps.Equal(got, want) { + t.Fatalf("session-end archives = %v, want %v", got, want) + } +} + // A Runner is shared across tenants and its door carries none. A lost session of a // non-bootstrap tenant, read cold from the table, must still be released durably and // woken: the hub resolves the session's tenant before it reads or deletes. From 239833859d89fa7b274391622067da8480ff9ace Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 4 Oct 2026 09:27:10 -0400 Subject: [PATCH 2/3] test(runnerhub): cover the enroll sweep-fault archive tenant (RIG-4296) Co-authored-by: Matt Wilkinson --- go/internal/runnerhub/lost_session_test.go | 42 +++++++++++++++++++ .../runnerhub/runner_tenant_pgtest_test.go | 20 +-------- 2 files changed, 43 insertions(+), 19 deletions(-) diff --git a/go/internal/runnerhub/lost_session_test.go b/go/internal/runnerhub/lost_session_test.go index 806e1c540..1d4143932 100644 --- a/go/internal/runnerhub/lost_session_test.go +++ b/go/internal/runnerhub/lost_session_test.go @@ -4,6 +4,7 @@ package runnerhub import ( "context" + "errors" "testing" "time" @@ -62,6 +63,47 @@ func recvEnded(t *testing.T, c chanEndSink) string { } } +type endedArchive struct { + tenant store.TenantID + sessionID string +} + +type tenantRecordingEndSink struct{ ended chan endedArchive } + +func (s tenantRecordingEndSink) OnSessionEnded(ctx context.Context, sessionID string) { + tenant, _ := store.TenantFromContext(ctx) + s.ended <- endedArchive{tenant: tenant, sessionID: sessionID} +} + +// When the durable sweep faults, the RAM-snapshot fallback still archives each +// session under its own tenant, read from the binding rows the fault left behind. +func TestEnrollReapFaultArchivesUnderSessionTenant(t *testing.T) { + ctx := t.Context() + hub := newHubOnly() + bindings := newFakeBindingStore() + hub.SetSessionBindingStore(bindings) + ended := tenantRecordingEndSink{ended: make(chan endedArchive, 1)} + hub.SetSessionEndSink(ended) + subj := runnerSubject() + tier, egress := compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED + hub.enroll(ctx, "runner-1", subj, tier, egress) + hub.bindContainer("cont-1", testAgentAccount, "runner-1") + hub.promoteSession(ctx, "cont-1", "sess-1") + + bindings.mu.Lock() + bindings.deleteForRunnerErr = errors.New("sweep fault") + bindings.mu.Unlock() + hub.enroll(ctx, "runner-1", subj, tier, egress) + select { + case got := <-ended.ended: + if got != (endedArchive{tenant: bindings.tenant, sessionID: "sess-1"}) { + t.Fatalf("archived %+v, want sess-1 under %q", got, bindings.tenant) + } + case <-time.After(10 * time.Second): + t.Fatal("no session-end archive within 10s") + } +} + // A session that ends without Stop is archived: on a lost-session drop by its owning // Runner, and on each binding the re-enroll reap removes. A foreign refusal is not. func TestSessionEndedWithoutStopIsArchived(t *testing.T) { diff --git a/go/internal/runnerhub/runner_tenant_pgtest_test.go b/go/internal/runnerhub/runner_tenant_pgtest_test.go index 894cfaf3b..fdfc71d46 100644 --- a/go/internal/runnerhub/runner_tenant_pgtest_test.go +++ b/go/internal/runnerhub/runner_tenant_pgtest_test.go @@ -16,21 +16,6 @@ import ( "github.com/RigelBuild/compass/go/internal/store" ) -type tenantRecordingEndSink struct { - ended chan struct { - tenant store.TenantID - sessionID string - } -} - -func (s tenantRecordingEndSink) OnSessionEnded(ctx context.Context, sessionID string) { - tenant, _ := store.TenantFromContext(ctx) - s.ended <- struct { - tenant store.TenantID - sessionID string - }{tenant: tenant, sessionID: sessionID} -} - func TestEnrollReapsEveryTenantsBindings(t *testing.T) { ctx := context.Background() dsn := pgtest.RequireDSN(t) @@ -66,10 +51,7 @@ func TestEnrollReapsEveryTenantsBindings(t *testing.T) { hub := newHubOnly() hub.SetSessionBindingStore(st) - ended := tenantRecordingEndSink{ended: make(chan struct { - tenant store.TenantID - sessionID string - }, 2)} + ended := tenantRecordingEndSink{ended: make(chan endedArchive, 2)} hub.SetSessionEndSink(ended) hub.enroll(ctx, "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) From b573c0732d73f46093d1d074c3f3c6f36fdf4cc2 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 4 Oct 2026 12:49:28 -0400 Subject: [PATCH 3/3] test(runnerhub): pin the ambiguous-tenant skip; fold the reap archive check into the cross-tenant test (RIG-4296) Co-authored-by: Matt Wilkinson --- go/internal/runnerhub/binding_cache_test.go | 4 + go/internal/runnerhub/hub.go | 3 +- go/internal/runnerhub/lost_session_test.go | 31 +++++++ .../runnerhub/runner_tenant_pgtest_test.go | 82 +++---------------- 4 files changed, 49 insertions(+), 71 deletions(-) diff --git a/go/internal/runnerhub/binding_cache_test.go b/go/internal/runnerhub/binding_cache_test.go index fec4fb581..0feecff98 100644 --- a/go/internal/runnerhub/binding_cache_test.go +++ b/go/internal/runnerhub/binding_cache_test.go @@ -58,6 +58,7 @@ type fakeBindingStore struct { resolveErr error reverseErr error deleteForRunnerErr error + tenantErr error } func newFakeBindingStore() *fakeBindingStore { @@ -156,6 +157,9 @@ const testRunnerID = "runner-1" func (f *fakeBindingStore) SessionBindingTenant(_ context.Context, sessionID, runnerID string) (store.TenantID, error) { f.mu.Lock() defer f.mu.Unlock() + if f.tenantErr != nil { + return "", f.tenantErr + } if b, ok := f.bindings[sessionID]; ok && b.RunnerID == runnerID { return f.tenant, nil } diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index 6aa434548..2b8bee947 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -1073,7 +1073,8 @@ func (h *Hub) enroll(ctx context.Context, id string, subject store.Subject, tier h.archiveEnded(store.WithTenant(ctx, b.TenantID), b.SessionID) } } else { - // The failed sweep left the rows, so each session's tenant is still readable. + // A failed sweep leaves the rows: resolve each session's tenant from them when + // the store still answers, else archive unscoped as before. An ambiguous id is skipped. for _, sessionID := range reapedSessions { if sctx, ok := h.runnerSessionCtx(ctx, id, sessionID); ok { h.archiveEnded(sctx, sessionID) diff --git a/go/internal/runnerhub/lost_session_test.go b/go/internal/runnerhub/lost_session_test.go index 1d4143932..63030dace 100644 --- a/go/internal/runnerhub/lost_session_test.go +++ b/go/internal/runnerhub/lost_session_test.go @@ -6,6 +6,7 @@ import ( "context" "errors" "testing" + "testing/synctest" "time" compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" @@ -104,6 +105,36 @@ func TestEnrollReapFaultArchivesUnderSessionTenant(t *testing.T) { } } +// A session id bound in several tenants resolves to no single tenant, so the +// fallback skips its archive rather than reading an arbitrary tenant's copy. +func TestEnrollReapFaultSkipsAmbiguousSessionArchive(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx := t.Context() + hub := newHubOnly() + bindings := newFakeBindingStore() + hub.SetSessionBindingStore(bindings) + ended := tenantRecordingEndSink{ended: make(chan endedArchive, 1)} + hub.SetSessionEndSink(ended) + subj := runnerSubject() + tier, egress := compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED + hub.enroll(ctx, "runner-1", subj, tier, egress) + hub.bindContainer("cont-1", testAgentAccount, "runner-1") + hub.promoteSession(ctx, "cont-1", "sess-1") + + bindings.mu.Lock() + bindings.deleteForRunnerErr = errors.New("sweep fault") + bindings.tenantErr = store.ErrConflict + bindings.mu.Unlock() + hub.enroll(ctx, "runner-1", subj, tier, egress) + synctest.Wait() + select { + case got := <-ended.ended: + t.Fatalf("archived %+v for an ambiguous session id, want no archive", got) + default: + } + }) +} + // A session that ends without Stop is archived: on a lost-session drop by its owning // Runner, and on each binding the re-enroll reap removes. A foreign refusal is not. func TestSessionEndedWithoutStopIsArchived(t *testing.T) { diff --git a/go/internal/runnerhub/runner_tenant_pgtest_test.go b/go/internal/runnerhub/runner_tenant_pgtest_test.go index fdfc71d46..e36b14455 100644 --- a/go/internal/runnerhub/runner_tenant_pgtest_test.go +++ b/go/internal/runnerhub/runner_tenant_pgtest_test.go @@ -4,8 +4,6 @@ package runnerhub import ( "context" - "errors" - "maps" "testing" "time" @@ -16,73 +14,6 @@ import ( "github.com/RigelBuild/compass/go/internal/store" ) -func TestEnrollReapsEveryTenantsBindings(t *testing.T) { - ctx := context.Background() - dsn := pgtest.RequireDSN(t) - st, err := store.Open(ctx, dsn) - if err != nil { - t.Fatalf("store.Open: %v", err) - } - t.Cleanup(st.Close) - bootstrap, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: "admin", DisplayName: "admin"}) - if err != nil { - t.Fatalf("BootstrapAdmin: %v", err) - } - bootstrapAgent, err := st.CreateAgent(ctx, bootstrap.ID, store.NewAgent{Handle: "agent-bootstrap", DisplayName: "agent-bootstrap"}) - if err != nil { - t.Fatalf("CreateAgent bootstrap: %v", err) - } - tenantB := seedTenantRow(t, ctx, dsn, "tenant-b-reap") - ctxB := store.WithTenant(ctx, tenantB) - owner, err := st.CreateUser(ctxB, store.NewUser{Handle: "owner-b-reap", DisplayName: "owner-b-reap"}) - if err != nil { - t.Fatalf("CreateUser: %v", err) - } - agent, err := st.CreateAgent(ctxB, owner.ID, store.NewAgent{Handle: "agent-b-reap", DisplayName: "agent-b-reap"}) - if err != nil { - t.Fatalf("CreateAgent: %v", err) - } - if _, err := st.RecordSessionBinding(ctxB, "sess-b", agent.ID, "runner-1"); err != nil { - t.Fatalf("RecordSessionBinding tenant B: %v", err) - } - if _, err := st.RecordSessionBinding(ctx, "sess-bootstrap", bootstrapAgent.ID, "runner-1"); err != nil { - t.Fatalf("RecordSessionBinding bootstrap: %v", err) - } - - hub := newHubOnly() - hub.SetSessionBindingStore(st) - ended := tenantRecordingEndSink{ended: make(chan endedArchive, 2)} - hub.SetSessionEndSink(ended) - hub.enroll(ctx, "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) - - if _, _, err := st.ResolveSessionBinding(ctxB, "sess-b"); !errors.Is(err, store.ErrNotFound) { - t.Fatalf("tenant B binding resolve error = %v, want ErrNotFound", err) - } - if _, _, err := st.ResolveSessionBinding(ctx, "sess-bootstrap"); !errors.Is(err, store.ErrNotFound) { - t.Fatalf("bootstrap binding resolve error = %v, want ErrNotFound", err) - } - bootstrapTenant, err := st.BootstrapTenant(ctx) - if err != nil { - t.Fatalf("BootstrapTenant: %v", err) - } - want := map[string]store.TenantID{"sess-b": tenantB, "sess-bootstrap": bootstrapTenant} - got := make(map[string]store.TenantID, len(want)) - for range len(want) { - select { - case e := <-ended.ended: - if _, dup := got[e.sessionID]; dup { - t.Fatalf("session %q archived twice", e.sessionID) - } - got[e.sessionID] = e.tenant - case <-time.After(10 * time.Second): - t.Fatalf("session-end archives = %v, want %v", got, want) - } - } - if !maps.Equal(got, want) { - t.Fatalf("session-end archives = %v, want %v", got, want) - } -} - // A Runner is shared across tenants and its door carries none. A lost session of a // non-bootstrap tenant, read cold from the table, must still be released durably and // woken: the hub resolves the session's tenant before it reads or deletes. @@ -112,13 +43,16 @@ func TestDropLostSessionScopesToTheSessionTenant(t *testing.T) { } // An enrolling Runner spans tenants, so its system-role sweep must reap tenant B's -// binding and record its end event with tenant B's id. +// binding, record its end event with tenant B's id, and archive it under tenant B. func TestEnrollReapsSessionBindingAcrossTenants(t *testing.T) { ctx := context.Background() st, agent, ctxB := openTenantBSession(t, ctx) + tenantB, _ := store.TenantFromContext(ctxB) hub := newHubOnly() hub.SetSessionBindingStore(st) + ended := tenantRecordingEndSink{ended: make(chan endedArchive, 1)} + hub.SetSessionEndSink(ended) hub.enroll(ctx, "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) var remaining, ends int @@ -134,6 +68,14 @@ func TestEnrollReapsSessionBindingAcrossTenants(t *testing.T) { if remaining != 0 || ends != 1 { t.Fatalf("tenant B binding remaining = %d, end events = %d; want reaped binding and one end", remaining, ends) } + select { + case got := <-ended.ended: + if got != (endedArchive{tenant: tenantB, sessionID: "sess-b"}) { + t.Fatalf("archived %+v, want sess-b under %q", got, tenantB) + } + case <-time.After(10 * time.Second): + t.Fatal("no session-end archive within 10s") + } } // openTenantBSession opens a store with a non-bootstrap tenant B whose agent holds