perf(graph): cache merged SDL and SchemaUpdate per ref #841

Merged
argoyle merged 1 commits from precompute-schema-cache into main 2026-05-19 07:51:51 +00:00
2 changed files with 100 additions and 3 deletions
+74 -2
View File
@@ -10,6 +10,7 @@ import (
"gitlab.com/unboundsoftware/eventsourced/eventsourced"
"gitea.unbound.se/unboundsoftware/schemas/domain"
"gitea.unbound.se/unboundsoftware/schemas/graph/model"
"gitea.unbound.se/unboundsoftware/schemas/hash"
)
@@ -21,9 +22,19 @@ type Cache struct {
services map[string]map[string]map[string]struct{}
subGraphs map[string]string
lastUpdate map[string]string
mergedSDLs map[string]*mergedSDLEntry
schemaUpdates map[string]*model.SchemaUpdate
logger *slog.Logger
}
// mergedSDLEntry holds a precomputed merged SDL together with the lastUpdate
// id it was computed against, so stale entries can be detected on read.
type mergedSDLEntry struct {
ID string
SDL string
SubGraphs []*model.SubGraph
}
func (c *Cache) OrganizationByAPIKey(apiKey string) *domain.Organization {
c.mu.RLock()
defer c.mu.RUnlock()
@@ -184,8 +195,11 @@ func (c *Cache) Update(msg any, _ goamqp.Headers) (any, error) {
for service := range refs[ref] {
delete(c.subGraphs, subGraphKey(orgId, ref, service))
}
// Remove lastUpdate for this org/ref
delete(c.lastUpdate, refKey(orgId, ref))
// Remove cached results for this org/ref
rk := refKey(orgId, ref)
delete(c.lastUpdate, rk)
delete(c.mergedSDLs, rk)
delete(c.schemaUpdates, rk)
}
delete(c.services, orgId)
}
@@ -253,10 +267,68 @@ func New(logger *slog.Logger) *Cache {
services: make(map[string]map[string]map[string]struct{}),
subGraphs: make(map[string]string),
lastUpdate: make(map[string]string),
mergedSDLs: make(map[string]*mergedSDLEntry),
schemaUpdates: make(map[string]*model.SchemaUpdate),
logger: logger,
}
}
// GetMergedSDL returns the cached merged SDL for (orgId, ref) if it was
// computed against the current lastUpdate. Returns nil when missing or stale.
func (c *Cache) GetMergedSDL(orgId, ref string) *mergedSDLEntry {
c.mu.RLock()
defer c.mu.RUnlock()
key := refKey(orgId, ref)
entry := c.mergedSDLs[key]
if entry == nil || entry.ID != c.lastUpdate[key] {
return nil
}
return entry
}
// MergedSDLEntry exposes the cached merged SDL fields to callers.
func (e *mergedSDLEntry) Unpack() (id, sdl string, subGraphs []*model.SubGraph) {
return e.ID, e.SDL, e.SubGraphs
}
// SetMergedSDL stores a precomputed merged SDL for (orgId, ref). The entry
// is only retained while c.lastUpdate[key] matches id; subsequent updates
// invalidate it implicitly via the version mismatch in GetMergedSDL.
func (c *Cache) SetMergedSDL(orgId, ref, id, sdl string, subGraphs []*model.SubGraph) {
c.mu.Lock()
defer c.mu.Unlock()
c.mergedSDLs[refKey(orgId, ref)] = &mergedSDLEntry{
ID: id,
SDL: sdl,
SubGraphs: subGraphs,
}
}
// GetSchemaUpdate returns the cached SchemaUpdate (subgraphs + cosmo router
// config) for (orgId, ref) when its id matches the current lastUpdate.
// Returns nil when missing or stale.
func (c *Cache) GetSchemaUpdate(orgId, ref string) *model.SchemaUpdate {
c.mu.RLock()
defer c.mu.RUnlock()
key := refKey(orgId, ref)
upd := c.schemaUpdates[key]
if upd == nil || upd.ID != c.lastUpdate[key] {
return nil
}
return upd
}
// SetSchemaUpdate stores a precomputed SchemaUpdate for (orgId, ref).
func (c *Cache) SetSchemaUpdate(orgId, ref string, update *model.SchemaUpdate) {
c.mu.Lock()
defer c.mu.Unlock()
c.schemaUpdates[refKey(orgId, ref)] = update
}
func refKey(orgId string, ref string) string {
return fmt.Sprintf("%s<->%s", orgId, ref)
}
+26 -1
View File
@@ -210,6 +210,7 @@ func (r *mutationResolver) UpdateSubGraph(ctx context.Context, input model.Input
SubGraphs: subGraphs,
CosmoRouterConfig: &cosmoConfig,
}
r.Cache.SetSchemaUpdate(orgId, input.Ref, update)
r.Logger.Info(
"Publishing schema update to subscribers",
@@ -280,13 +281,25 @@ func (r *queryResolver) Supergraph(ctx context.Context, ref string, isAfter *str
if isAfter != nil {
after = *isAfter
}
services, lastUpdate := r.Cache.Services(orgId, ref, after)
_, lastUpdate := r.Cache.Services(orgId, ref, after)
if after == lastUpdate {
return &model.Unchanged{
ID: lastUpdate,
MinDelaySeconds: 10,
}, nil
}
if cached := r.Cache.GetMergedSDL(orgId, ref); cached != nil {
id, sdl, subGraphs := cached.Unpack()
return &model.SubGraphs{
ID: id,
SubGraphs: subGraphs,
Sdl: sdl,
MinDelaySeconds: 10,
}, nil
}
services, _ := r.Cache.Services(orgId, ref, "")
subGraphs := make([]*model.SubGraph, len(services))
serviceSDLs := make([]string, len(services))
for i, id := range services {
@@ -302,6 +315,7 @@ func (r *queryResolver) Supergraph(ctx context.Context, ref string, isAfter *str
if err != nil {
return nil, err
}
r.Cache.SetMergedSDL(orgId, ref, lastUpdate, sdl, subGraphs)
return &model.SubGraphs{
ID: lastUpdate,
SubGraphs: subGraphs,
@@ -344,6 +358,16 @@ func (r *queryResolver) LatestSchema(ctx context.Context, ref string) (*model.Sc
return nil, fmt.Errorf("no authentication provided")
}
if cached := r.Cache.GetSchemaUpdate(orgId, ref); cached != nil {
r.Logger.Info(
"Latest schema served from cache",
"ref", ref,
"orgId", orgId,
"id", cached.ID,
)
return cached, nil
}
// Get current services and schema
services, lastUpdate := r.Cache.Services(orgId, ref, "")
r.Logger.Info(
@@ -385,6 +409,7 @@ func (r *queryResolver) LatestSchema(ctx context.Context, ref string) (*model.Sc
SubGraphs: subGraphs,
CosmoRouterConfig: &cosmoConfig,
}
r.Cache.SetSchemaUpdate(orgId, ref, update)
r.Logger.Info(
"Latest schema fetched",