diff --git a/internal/api/api.go b/internal/api/api.go index 0006d4d..14e760d 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -37,6 +37,9 @@ type Server struct { // pointer so a test can wire the prober to a stub upstream; when nil, // handleReadyz builds one from the server's own clients. Health *HealthProber + // RegistryStaleness reads the configured attested set for /readyz. Its + // result is informational and does not affect scan readiness. + RegistryStaleness *RegistryStalenessProber // Metrics holds the scan outcome and per-source failure counters served // at GET /metrics. Nil-safe: the handlers treat a nil Metrics as // "nothing counted", never as a crash. @@ -76,6 +79,12 @@ var Routes = []string{ // per-request logger — the correlation the outcome log line depends on is // installed once, here, rather than remembered per handler. func (s *Server) Handler() http.Handler { + // Build the prober once after callers have had a chance to configure the + // scanner and registry reader. Rebuilding it per request would bypass its + // cache and turn /readyz scrapes into upstream and RPC traffic. + if s.Health == nil { + s.Health = NewHealthProber(s) + } mux := http.NewServeMux() // Exposure is one request to many scans, so it is wrapped in the rate // limit at registration — the bucket is built once per Handler, not per diff --git a/internal/api/health.go b/internal/api/health.go index 321d5e8..0b93d6f 100644 --- a/internal/api/health.go +++ b/internal/api/health.go @@ -2,11 +2,14 @@ package api import ( "context" + "errors" "fmt" "io" "net/http" "sync" "time" + + "github.com/use-assay/assay/internal/registry" ) // Everything in this file answers one question — "can Assay do its job right @@ -143,6 +146,9 @@ type HealthProber struct { // upstreams is the set of upstreams this prober reports on, captured when // the prober is built. upstreams []upstream + // Registry checks the configured attested set. Its result is informational + // and never changes readiness status. + Registry *RegistryStalenessProber mu sync.Mutex last time.Time // start of the most recent real probe round @@ -160,7 +166,10 @@ type probeFlight struct { // NewHealthProber returns a prober wired to the same upstreams the server's // scanner reads from. func NewHealthProber(s *Server) *HealthProber { - return &HealthProber{upstreams: probeEndpoints(s)} + return &HealthProber{ + upstreams: probeEndpoints(s), + Registry: s.RegistryStaleness, + } } // now is the prober's clock, overridable in tests. @@ -191,6 +200,149 @@ type Readiness struct { // Upstreams reports one entry per checkable source, in the fixed order // the prober was built with, so a diff between two answers is stable. Upstreams []UpstreamStatus `json:"upstreams"` + // RegistryStaleness is informational. A stale or unavailable registry + // read never changes the readiness status. + RegistryStaleness RegistryStaleness `json:"registry_staleness"` +} + +// RegistrySafetyReader reads one attestation from the configured registry. +type RegistrySafetyReader interface { + GetSafety(context.Context, string) (*registry.Safety, error) +} + +// RegistryAsset identifies one configured attested-set member and its +// network-specific Stellar Asset Contract address. +type RegistryAsset struct { + Asset string + SAC string +} + +// RegistryStalenessProber reads the configured attested set. The enclosing +// HealthProber caches its result with the rest of the readiness round. +type RegistryStalenessProber struct { + Reader RegistrySafetyReader + Assets []RegistryAsset + Window time.Duration + // ConfigurationError keeps a bad or unavailable local config visible as + // unknown without making the scanner unavailable. + ConfigurationError string +} + +// RegistryStaleness describes freshness of the configured on-chain set. +type RegistryStaleness struct { + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + CheckedAt time.Time `json:"checked_at"` + WindowSecs int64 `json:"window_secs"` + AssetCount int `json:"asset_count"` + AttestedCount int `json:"attested_count"` + UnattestedCount int `json:"unattested_count"` + StaleCount int `json:"stale_count"` + OldestAttestation *OldestAttestation `json:"oldest_attestation,omitempty"` +} + +// OldestAttestation is the oldest present entry in the configured set. +type OldestAttestation struct { + Asset string `json:"asset"` + AttestedAt time.Time `json:"attested_at"` + AgeSecs int64 `json:"age_secs"` +} + +// Check reads every configured set member and compares attested_at with the +// configured window. Any RPC failure makes the aggregate unknown rather than +// presenting partial counts as a complete set. +func (p *RegistryStalenessProber) Check(ctx context.Context, now time.Time) RegistryStaleness { + windowSecs := int64(0) + assetCount := 0 + if p != nil && p.Window > 0 { + windowSecs = int64(p.Window.Seconds()) + if p.Window%time.Second != 0 { + windowSecs++ + } + } + if p != nil { + assetCount = len(p.Assets) + } + unknown := func(reason string) RegistryStaleness { + return RegistryStaleness{ + Status: "unknown", + Reason: reason, + CheckedAt: now, + WindowSecs: windowSecs, + AssetCount: assetCount, + } + } + if p == nil { + return unknown("registry staleness is not configured") + } + if p.ConfigurationError != "" { + return unknown(p.ConfigurationError) + } + if p.Reader == nil || len(p.Assets) == 0 || p.Window <= 0 { + return unknown("registry staleness is not configured") + } + + type result struct { + safety *registry.Safety + err error + } + results := make([]result, len(p.Assets)) + var wg sync.WaitGroup + for i, asset := range p.Assets { + wg.Add(1) + go func(i int, asset RegistryAsset) { + defer wg.Done() + results[i].safety, results[i].err = p.Reader.GetSafety(ctx, asset.SAC) + }(i, asset) + } + wg.Wait() + + status := RegistryStaleness{ + Status: "current", + CheckedAt: now, + WindowSecs: windowSecs, + AssetCount: len(p.Assets), + } + const maxUnixTimestamp = uint64(1<<63 - 1) + for i, res := range results { + switch { + case errors.Is(res.err, registry.ErrNotAttested): + status.UnattestedCount++ + continue + case res.err != nil: + return unknown(fmt.Sprintf("registry RPC read failed for %s", p.Assets[i].Asset)) + case res.safety == nil: + return unknown(fmt.Sprintf("registry returned no attestation result for %s", p.Assets[i].Asset)) + case res.safety.AttestedAt > maxUnixTimestamp: + return unknown(fmt.Sprintf("registry returned an invalid attested_at for %s", p.Assets[i].Asset)) + } + + attestedAt := time.Unix(int64(res.safety.AttestedAt), 0).UTC() + age := now.Sub(attestedAt) + if age < 0 { + age = 0 + } + status.AttestedCount++ + if age > p.Window { + status.StaleCount++ + } + if status.OldestAttestation == nil || + attestedAt.Before(status.OldestAttestation.AttestedAt) { + status.OldestAttestation = &OldestAttestation{ + Asset: p.Assets[i].Asset, + AttestedAt: attestedAt, + AgeSecs: int64(age.Seconds()), + } + } + } + + switch { + case status.StaleCount > 0: + status.Status = "stale" + case status.UnattestedCount > 0: + status.Status = "incomplete" + } + return status } // UpstreamStatus is one source's reachability as the readiness round measured @@ -277,7 +429,7 @@ func (p *HealthProber) Run(ctx context.Context) Readiness { case <-ctx.Done(): // The caller gave up. Nothing is reported — in particular not // healthy. The next caller runs the round. - return Readiness{Status: readyStateUnknown, CheckedAt: now} + return unknownReadiness(now) } } @@ -327,6 +479,7 @@ func stampAge(ans *Readiness, now time.Time) { func (p *HealthProber) probeAll(ctx context.Context, now time.Time) Readiness { ups := p.upstreams results := make([]UpstreamStatus, len(ups)) + var registryStatus RegistryStaleness client := p.HTTP if client == nil { @@ -334,6 +487,11 @@ func (p *HealthProber) probeAll(ctx context.Context, now time.Time) Readiness { } var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + registryStatus = p.Registry.Check(ctx, now) + }() for i, u := range ups { wg.Add(1) go func(i int, u upstream) { @@ -358,8 +516,9 @@ func (p *HealthProber) probeAll(ctx context.Context, now time.Time) Readiness { } ans := Readiness{ - CheckedAt: now, - Upstreams: results, + CheckedAt: now, + Upstreams: results, + RegistryStaleness: registryStatus, } switch { case horizonDown: @@ -422,9 +581,10 @@ func (p *HealthProber) probeOne(ctx context.Context, client *http.Client, u upst // never healthy. func unknownReadiness(now time.Time) Readiness { return Readiness{ - Status: readyStateUnknown, - Reason: "the readiness round could not be observed; reported as unknown rather than as healthy", - CheckedAt: now, + Status: readyStateUnknown, + Reason: "the readiness round could not be observed; reported as unknown rather than as healthy", + CheckedAt: now, + RegistryStaleness: RegistryStaleness{Status: "unknown", Reason: "readiness round could not be observed", CheckedAt: now}, } } diff --git a/internal/api/health_test.go b/internal/api/health_test.go index 109a61a..71ac01d 100644 --- a/internal/api/health_test.go +++ b/internal/api/health_test.go @@ -3,16 +3,19 @@ package api_test import ( "context" "encoding/json" + "errors" "io" "log/slog" "net/http" "net/http/httptest" + "strings" "sync" "sync/atomic" "testing" "time" "github.com/use-assay/assay/internal/api" + "github.com/use-assay/assay/internal/registry" ) // newProbeServer returns a Server whose scanner is wired to the given stub @@ -79,11 +82,52 @@ func (s *upstreamStub) ServeHTTP(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("ok")) } +type registryStub struct { + mu sync.Mutex + records map[string]*registry.Safety + errors map[string]error + calls int +} + +func (s *registryStub) GetSafety(_ context.Context, asset string) (*registry.Safety, error) { + s.mu.Lock() + s.calls++ + s.mu.Unlock() + if err := s.errors[asset]; err != nil { + return nil, err + } + record, ok := s.records[asset] + if !ok { + return nil, registry.ErrNotAttested + } + return record, nil +} + +func (s *registryStub) Calls() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.calls +} + // readyBody is the decoded /readyz response. type readyBody struct { - Status string `json:"status"` - Reason string `json:"reason"` - AgeSecs int64 `json:"age_secs"` + Status string `json:"status"` + Reason string `json:"reason"` + AgeSecs int64 `json:"age_secs"` + Registry struct { + Status string `json:"status"` + Reason string `json:"reason"` + WindowSecs int64 `json:"window_secs"` + AssetCount int `json:"asset_count"` + AttestedCount int `json:"attested_count"` + UnattestedCount int `json:"unattested_count"` + StaleCount int `json:"stale_count"` + Oldest *struct { + Asset string `json:"asset"` + AttestedAt time.Time `json:"attested_at"` + AgeSecs int64 `json:"age_secs"` + } `json:"oldest_attestation"` + } `json:"registry_staleness"` Upstreams []struct { Name string `json:"name"` Kind string `json:"kind"` @@ -93,6 +137,88 @@ type readyBody struct { } `json:"upstreams"` } +func TestReadyzReportsAndCachesRegistryStaleness(t *testing.T) { + now := time.Date(2026, 10, 23, 6, 0, 0, 0, time.UTC) + horizon := httptest.NewServer(newUpstreamStub()) + defer horizon.Close() + expert := httptest.NewServer(newUpstreamStub()) + defer expert.Close() + reader := ®istryStub{ + records: map[string]*registry.Safety{ + "sac-fresh": {AttestedAt: uint64(now.Add(-24 * time.Hour).Unix())}, + "sac-stale": {AttestedAt: uint64(now.Add(-25 * time.Hour).Unix())}, + }, + errors: map[string]error{}, + } + srv, _ := newProbeServer(horizon.URL, expert.URL) + srv.RegistryStaleness = &api.RegistryStalenessProber{ + Reader: reader, + Assets: []api.RegistryAsset{ + {Asset: "FRESH-G...", SAC: "sac-fresh"}, + {Asset: "STALE-G...", SAC: "sac-stale"}, + {Asset: "MISSING-G...", SAC: "sac-missing"}, + }, + Window: 24 * time.Hour, + } + srv.Health = api.NewHealthProber(srv) + srv.Health.Now = func() time.Time { return now } + h := srv.Handler() + + _, body := getReadyz(h) + if body.Registry.Status != "stale" { + t.Fatalf("registry status = %q, want stale", body.Registry.Status) + } + if body.Registry.WindowSecs != 24*60*60 || + body.Registry.AssetCount != 3 || + body.Registry.AttestedCount != 2 || + body.Registry.UnattestedCount != 1 || + body.Registry.StaleCount != 1 { + t.Fatalf("unexpected registry counts: %+v", body.Registry) + } + if body.Registry.Oldest == nil || + body.Registry.Oldest.Asset != "STALE-G..." || + body.Registry.Oldest.AgeSecs != 25*60*60 { + t.Fatalf("oldest attestation = %+v", body.Registry.Oldest) + } + + _, cached := getReadyz(h) + if cached.Registry.Status != "stale" || reader.Calls() != 3 { + t.Fatalf("cached registry status = %q; RPC reads = %d, want stale and 3", + cached.Registry.Status, reader.Calls()) + } +} + +func TestReadyzReportsRegistryRPCFailureAsUnknown(t *testing.T) { + horizon := httptest.NewServer(newUpstreamStub()) + defer horizon.Close() + expert := httptest.NewServer(newUpstreamStub()) + defer expert.Close() + reader := ®istryStub{ + records: map[string]*registry.Safety{}, + errors: map[string]error{"sac-fail": errors.New("provider credential must not be exposed")}, + } + srv, _ := newProbeServer(horizon.URL, expert.URL) + srv.RegistryStaleness = &api.RegistryStalenessProber{ + Reader: reader, + Assets: []api.RegistryAsset{{Asset: "USDC-G...", SAC: "sac-fail"}}, + Window: 24 * time.Hour, + } + srv.Health = api.NewHealthProber(srv) + h := srv.Handler() + + rec, body := getReadyz(h) + if rec.Code != http.StatusOK || body.Status != "ready" { + t.Fatalf("readiness = %d %q, want 200 ready despite registry outage", rec.Code, body.Status) + } + if body.Registry.Status != "unknown" || + body.Registry.Reason != "registry RPC read failed for USDC-G..." { + t.Fatalf("registry failure status = %+v", body.Registry) + } + if strings.Contains(rec.Body.String(), "credential") { + t.Fatal("registry error response exposed provider credential detail") + } +} + func getReadyz(h http.Handler) (*httptest.ResponseRecorder, readyBody) { rec := httptest.NewRecorder() h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/readyz", nil)) diff --git a/openapi.yaml b/openapi.yaml index c1343d3..084ac38 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -553,6 +553,40 @@ components: type: array items: $ref: '#/components/schemas/UpstreamStatus' + registry_staleness: + type: object + description: Informational registry freshness; does not affect readiness status. + required: [status, checked_at, window_secs, asset_count, attested_count, unattested_count, stale_count] + properties: + status: + type: string + enum: [current, stale, incomplete, unknown] + reason: + type: string + checked_at: + type: string + format: date-time + window_secs: + type: integer + asset_count: + type: integer + attested_count: + type: integer + unattested_count: + type: integer + stale_count: + type: integer + oldest_attestation: + type: object + required: [asset, attested_at, age_secs] + properties: + asset: + type: string + attested_at: + type: string + format: date-time + age_secs: + type: integer UpstreamStatus: type: object required: [name, kind, reachable, probe_url]