package httpapi import ( "context" "crypto/ed25519" "crypto/sha256" "encoding/binary" "encoding/hex" "errors" "fmt" "sort" "strings" "sync" "evobgp/internal/bundle" "evobgp/internal/pipeline" "evobgp/internal/store" ) const ( // nodeBundleCacheMaxEntries is the maximum number of finished node bundles retained. nodeBundleCacheMaxEntries = 32 // nodeBundleCacheMaxBytes is the maximum total size of cached bundle bodies (64 MiB). nodeBundleCacheMaxBytes = 64 << 20 ) // errNodeBundleRevisionMissing is returned when the revision cannot be loaded. // It is not cached: the next pull tries again. var errNodeBundleRevisionMissing = errors.New("httpapi: revision not found") // nodeBundleOpError carries the writeInternalError operation name for a failed build. type nodeBundleOpError struct { op string err error } func (e *nodeBundleOpError) Error() string { if e == nil || e.err == nil { return "httpapi: node bundle" } return fmt.Sprintf("httpapi: %s: %v", e.op, e.err) } func (e *nodeBundleOpError) Unwrap() error { return e.err } // nodeBundleBuilder produces the gzip body served to evobgp-node. // The default implementation is bundle.BuildGzippedTar, whose archive contains // manifest.json and bundle.sig. Tests replace it to count rebuilds. type nodeBundleBuilder func(revisionID, speakerID string, fragments map[string]string, priv ed25519.PrivateKey) ([]byte, error) // nodeBundleCache stores finished gzip bodies. A hit returns the bytes of the // build that filled the entry, signature included. Errors are not stored. // Concurrent loads of one key share a single build. type nodeBundleCache struct { mu sync.Mutex maxN int maxBytes int total int items map[string]*nodeBundleEntry lru []string // index 0 is the oldest idle entry flights map[string]*nodeBundleFlight } type nodeBundleEntry struct { body []byte } type nodeBundleFlight struct { done chan struct{} body []byte err error joined int } func newNodeBundleCache(maxEntries, maxBytes int) *nodeBundleCache { return &nodeBundleCache{ maxN: maxEntries, maxBytes: maxBytes, items: make(map[string]*nodeBundleEntry), flights: make(map[string]*nodeBundleFlight), } } // nodeBundleCacheKey identifies one servable bundle. // overlayFP is a digest of the speaker overlay output (router id / source addresses, // node IPv4 fallback, peer and discovery render). Revision content itself is immutable, // so the revision id stands in for the shared prefix files. func nodeBundleCacheKey(tenantID, revisionID, speakerID, overlayFP string) string { var b strings.Builder b.Grow(len(tenantID) + len(revisionID) + len(speakerID) + len(overlayFP) + 3) b.WriteString(tenantID) b.WriteByte(0) b.WriteString(revisionID) b.WriteByte(0) b.WriteString(speakerID) b.WriteByte(0) b.WriteString(overlayFP) return b.String() } // nodeBundleOverlayFingerprint hashes the speaker overlay that handleNodeBundle // would apply on top of an immutable revision. Empty base fragments are enough: // bird.conf and the peers fragment are fully re-rendered from tenant settings, // speaker source addresses, and the live peer set. Revision file bytes are not // part of this digest. func nodeBundleOverlayFingerprint(st store.Backend, tenantID, speakerID, revisionID string) (string, error) { frags, err := pipeline.OverlayFragmentsForSpeaker(st, tenantID, speakerID, revisionID, nil) if err != nil { return "", fmt.Errorf("httpapi: bundle overlay: %w", err) } keys := make([]string, 0, len(frags)) for k := range frags { keys = append(keys, k) } sort.Strings(keys) h := sha256.New() for _, k := range keys { if err := hashWriteString(h, k); err != nil { return "", fmt.Errorf("httpapi: bundle overlay: %w", err) } if err := hashWriteString(h, frags[k]); err != nil { return "", fmt.Errorf("httpapi: bundle overlay: %w", err) } } return hex.EncodeToString(h.Sum(nil)), nil } func hashWriteString(h interface{ Write([]byte) (int, error) }, s string) error { var buf [8]byte binary.BigEndian.PutUint64(buf[:], uint64(len(s))) if _, err := h.Write(buf[:]); err != nil { return err } _, err := h.Write([]byte(s)) return err } // do returns the cached body or runs build once for key. // build's bool is false when the result must not be stored (failure is also not stored). // Callers must not mutate the returned slice. func (c *nodeBundleCache) do(ctx context.Context, key string, build func(context.Context) ([]byte, bool, error)) (body []byte, err error) { if c == nil { return nil, errors.New("httpapi: nil node bundle cache") } if ctx == nil { ctx = context.Background() } c.mu.Lock() if e, ok := c.items[key]; ok { c.touchLocked(key) body = e.body c.mu.Unlock() return body, nil } if f, ok := c.flights[key]; ok { f.joined++ c.mu.Unlock() return waitNodeBundleFlight(ctx, f) } if err = ctx.Err(); err != nil { c.mu.Unlock() return nil, err } f := &nodeBundleFlight{done: make(chan struct{})} c.flights[key] = f c.mu.Unlock() var cacheOK bool var finished bool finish := func() { if finished { return } finished = true out := body c.mu.Lock() delete(c.flights, key) if err == nil && cacheOK { if stored, ok := c.storeLocked(key, out); ok { out = stored } } f.body = out f.err = err close(f.done) c.mu.Unlock() body = out } defer func() { if rec := recover(); rec != nil { body = nil cacheOK = false err = fmt.Errorf("httpapi: node bundle build: %v", rec) } finish() }() // Detach from the leader's cancel so one disconnected agent does not fail the shared build. body, cacheOK, err = build(context.WithoutCancel(ctx)) return } func waitNodeBundleFlight(ctx context.Context, f *nodeBundleFlight) ([]byte, error) { select { case <-f.done: return f.body, f.err case <-ctx.Done(): select { case <-f.done: return f.body, f.err default: return nil, ctx.Err() } } } // storeLocked retains body when it fits under both caps, evicting the oldest idle // entries first. body is copied. The stored slice is returned. func (c *nodeBundleCache) storeLocked(key string, body []byte) ([]byte, bool) { n := len(body) if n == 0 || c.maxN <= 0 || c.maxBytes <= 0 || n > c.maxBytes { return nil, false } if e, ok := c.items[key]; ok { c.total -= len(e.body) delete(c.items, key) c.removeLRULocked(key) } for len(c.items) > 0 && (len(c.items) >= c.maxN || c.total+n > c.maxBytes) { if len(c.lru) == 0 { return nil, false } c.deleteOldestLocked() } if len(c.items) >= c.maxN || c.total+n > c.maxBytes { return nil, false } stored := make([]byte, n) copy(stored, body) c.items[key] = &nodeBundleEntry{body: stored} c.total += n c.lru = append(c.lru, key) return stored, true } func (c *nodeBundleCache) touchLocked(key string) { c.removeLRULocked(key) c.lru = append(c.lru, key) } func (c *nodeBundleCache) deleteOldestLocked() { if len(c.lru) == 0 { return } key := c.lru[0] c.lru = c.lru[1:] if e, ok := c.items[key]; ok { c.total -= len(e.body) delete(c.items, key) } } func (c *nodeBundleCache) removeLRULocked(key string) { for i, k := range c.lru { if k != key { continue } copy(c.lru[i:], c.lru[i+1:]) c.lru = c.lru[:len(c.lru)-1] return } } func (s *Server) buildNodeBundle(revisionID, speakerID string, fragments map[string]string) ([]byte, error) { if s != nil && s.nodeBundleBuild != nil { return s.nodeBundleBuild(revisionID, speakerID, fragments, s.bundlePriv) } return bundle.BuildGzippedTar(revisionID, speakerID, fragments, s.bundlePriv) } // loadNodeBundle reads the revision, applies the speaker overlay, and signs the gzip. // The bool is false when overlay inputs moved during the load, so the caller must not // cache the body under the key captured before the load. Failed loads are not cacheable. func (s *Server) loadNodeBundle(ctx context.Context, tenantID, speakerID, revisionID, overlayFP string) ([]byte, bool, error) { if err := ctx.Err(); err != nil { return nil, false, err } fpNow, err := nodeBundleOverlayFingerprint(s.store, tenantID, speakerID, revisionID) if err != nil { return nil, false, &nodeBundleOpError{op: "bundle overlay", err: err} } rev, err := s.store.GetRevision(tenantID, revisionID) if err != nil { return nil, false, errNodeBundleRevisionMissing } overlaid, err := pipeline.OverlayFragmentsForSpeaker(s.store, tenantID, speakerID, revisionID, rev.PreviewFragments) if err != nil { return nil, false, &nodeBundleOpError{op: "bundle overlay", err: err} } fpAfter, err := nodeBundleOverlayFingerprint(s.store, tenantID, speakerID, revisionID) if err != nil { return nil, false, &nodeBundleOpError{op: "bundle overlay", err: err} } body, err := s.buildNodeBundle(revisionID, speakerID, overlaid) if err != nil { return nil, false, &nodeBundleOpError{op: "internal", err: err} } cacheable := fpNow == overlayFP && fpAfter == fpNow && len(body) > 0 return body, cacheable, nil }