admin
fb248961c6
Manager: one-shot registration (marker gate; backoff cap 15m), adopt-lost-marker, re-key-on-mismatch, REVOKED-STAYS-REVOKED teardown (marker kept, zero execs on later ticks), no-teardown-on-absent-data, hash-gated apply (zero execs steady state), restart-not-reload on conf change, self-heal enable. Status stanza with latest-handshakes-ONLY wg read. Collector WireguardReporter seam. desired.Syncer AddConsumer fan-out with panic containment. Red-proofs a/b/d run + reverted; no-key-material-in-logs asserted. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
91 lines
2.9 KiB
Go
91 lines
2.9 KiB
Go
package desired
|
|
|
|
// S3 Group C — the raw-consumer fan-out seam: called on generation advance, NOT on no-advance,
|
|
// and a panicking consumer is contained (the guest reconcile path must never break).
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"sync"
|
|
"testing"
|
|
|
|
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
|
"gitea.dooplex.hu/admin/felhom-agent/internal/reconcile"
|
|
)
|
|
|
|
type recordingConsumer struct {
|
|
mu sync.Mutex
|
|
calls []*hub.DesiredStateResponse
|
|
panic bool
|
|
}
|
|
|
|
func (r *recordingConsumer) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse) {
|
|
r.mu.Lock()
|
|
r.calls = append(r.calls, resp)
|
|
r.mu.Unlock()
|
|
if r.panic {
|
|
panic("consumer exploded")
|
|
}
|
|
}
|
|
|
|
func (r *recordingConsumer) count() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.calls) }
|
|
|
|
type stubFetcher struct{ resp *hub.DesiredStateResponse }
|
|
|
|
func (s *stubFetcher) FetchDesiredState(context.Context) (*hub.DesiredStateResponse, error) {
|
|
return s.resp, nil
|
|
}
|
|
|
|
func testResp(gen int64) *hub.DesiredStateResponse {
|
|
return &hub.DesiredStateResponse{
|
|
Generation: gen,
|
|
DesiredState: hub.WireDesiredState{
|
|
Guests: []hub.WireDesiredGuest{},
|
|
Wireguard: &hub.WireWireguard{Pubkey: "PK", AssignedIP: "10.77.0.2/32"},
|
|
},
|
|
}
|
|
}
|
|
|
|
func TestSyncer_ConsumerCalledOnAdvanceOnly(t *testing.T) {
|
|
provider := reconcile.NewCachingProvider()
|
|
f := &stubFetcher{resp: testResp(2)}
|
|
s := NewSyncer(f, provider, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
c := &recordingConsumer{}
|
|
s.AddConsumer(c)
|
|
|
|
// Advance → fetch → consumer called with the raw doc (wireguard block intact).
|
|
s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 2})
|
|
if c.count() != 1 {
|
|
t.Fatalf("consumer calls = %d, want 1", c.count())
|
|
}
|
|
if c.calls[0].DesiredState.Wireguard == nil || c.calls[0].DesiredState.Wireguard.Pubkey != "PK" {
|
|
t.Fatalf("consumer got %+v — the raw wireguard block must ride through", c.calls[0].DesiredState.Wireguard)
|
|
}
|
|
|
|
// No advance → no fetch → no consumer call (the negative).
|
|
s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 2})
|
|
if c.count() != 1 {
|
|
t.Errorf("consumer called without a generation advance: %d", c.count())
|
|
}
|
|
}
|
|
|
|
func TestSyncer_PanickingConsumerContained(t *testing.T) {
|
|
provider := reconcile.NewCachingProvider()
|
|
f := &stubFetcher{resp: testResp(1)}
|
|
s := NewSyncer(f, provider, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
bomb := &recordingConsumer{panic: true}
|
|
after := &recordingConsumer{}
|
|
s.AddConsumer(bomb)
|
|
s.AddConsumer(after)
|
|
|
|
// Must not panic out; the second consumer still runs; the provider still updated.
|
|
s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 1})
|
|
if after.count() != 1 {
|
|
t.Errorf("consumer after the panicking one not called: %d", after.count())
|
|
}
|
|
if provider.Generation() != 1 {
|
|
t.Errorf("provider generation = %d, want 1 (guest path unaffected)", provider.Generation())
|
|
}
|
|
}
|