perf: 订阅 serve 缓存改为内容寻址(失效驱动,不再按 TTL 重跑引擎)

internal/server/server_plugin_invoke.go

@@ -269,6 +269,16 @@ func (s *Server) handlePluginCall(w http.ResponseWriter, r *http.Request, p prin
 		return
 	}
 	s.recordPluginCallAudit(p, req.ID, req.Service, req.Method, scopes, "allow", "")
+	// A successful mutation of a plugin's subscription store can change what
+	// shares sourcing it render. Drop those cached bodies now — otherwise the
+	// edit would only take effect at the cache's revalidation cadence, and the
+	// content hash cannot see it (the record changed, not the content).
+	if req.Service == req.ID+"/subscription" {
+		switch req.Method {
+		case "save", "delete", "import", "migrate":
+			s.invalidateSharesForPlugin(req.ID)
+		}
+	}
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
 	if len(out) == 0 {

internal/server/server_subscription_share.go

@@ -2,6 +2,8 @@ package server
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -78,12 +80,21 @@ const (
 	// bounds the classes per share, so this is a share-count budget rather than a
 	// defence against key explosion.
 	subscriptionCacheEntries = 512
-	// subscriptionCacheTTL keeps a body reusable for long enough to absorb a
-	// client's retry burst without making a rotation take noticeably longer to
-	// take effect. Rotation does not wait for it: it invalidates directly.
-	subscriptionCacheTTL = 300 * time.Second
+	// subscriptionCacheTTL is the revalidation cadence, not the freshness bound:
+	// an expired entry whose content hash still matches is extended without a
+	// re-render, so the engine only ever runs when the content actually moved.
+	// It aligns with subscriptionRefreshInterval so one poll cycle performs at
+	// most one provider fetch and zero renders in the steady state.
+	subscriptionCacheTTL = 30 * time.Minute
 )
 
+// subscriptionContentHash digests the render input. The bytes themselves are
+// never stored on the key; the hash is only ever compared for equality.
+func subscriptionContentHash(raw string) string {
+	sum := sha256.Sum256([]byte(raw))
+	return hex.EncodeToString(sum[:])
+}
+
 // handleSubscriptionShare serves the public subscription endpoint.
 //
 // The core owns every part of this - routing, lookup, rate limiting, audit and
@@ -141,6 +152,25 @@ func (s *Server) handleSubscriptionShare(w http.ResponseWriter, r *http.Request)
 	key := subscriptionCacheKey{ShareID: share.ID, Format: format, UAClass: uaClass}
 
 	body, contentType, userinfo, cached := s.subscriptionCache.Get(key, s.now())
+	if !cached && share.Source.Kind == model.ShareSourcePlugin {
+		// Revalidate before paying for a render. A render boots the plugin's
+		// JavaScript engine, which costs seconds; comparing the content digest
+		// costs a store read and, at most, one provider fetch. When the digest
+		// still matches, the cached body is exact and is extended. When the
+		// source cannot be reached at all, the last good body is served — a
+		// provider outage must not take a client's configuration with it, the
+		// same rule the snapshot layer applies one step down.
+		if stale, ok := s.subscriptionCache.GetStale(key); ok {
+			snap, snapErr := s.snapshotFor(r.Context(), share.Source.PluginID, share.Source.SubscriptionID, false)
+			switch {
+			case snapErr != nil:
+				body, contentType, userinfo, cached = stale.body, stale.contentType, stale.userinfo, true
+			case subscriptionContentHash(snap.Raw) == stale.contentHash:
+				s.subscriptionCache.Extend(key, s.now())
+				body, contentType, userinfo, cached = stale.body, stale.contentType, stale.userinfo, true
+			}
+		}
+	}
 	if !cached {
 		rendered, renderErr := s.renderShare(r.Context(), share, format, uaClass)
 		if renderErr != nil {
@@ -155,7 +185,15 @@ func (s *Server) handleSubscriptionShare(w http.ResponseWriter, r *http.Request)
 			deny("empty render refused", map[string]string{"slug": slug, "token_sha256": tokenHash, "share_id": share.ID})
 			return
 		}
-		s.subscriptionCache.Put(key, body, contentType, userinfo, s.now())
+		contentHash := ""
+		if share.Source.Kind == model.ShareSourcePlugin {
+			// renderShare already refreshed the snapshot, so this read is the
+			// fresh record, never a second fetch.
+			if snap, snapErr := s.snapshotFor(r.Context(), share.Source.PluginID, share.Source.SubscriptionID, false); snapErr == nil {
+				contentHash = subscriptionContentHash(snap.Raw)
+			}
+		}
+		s.subscriptionCache.Put(key, body, contentType, userinfo, contentHash, s.now())
 	}
 
 	if contentType == "" {
@@ -181,6 +219,30 @@ func (s *Server) handleSubscriptionShare(w http.ResponseWriter, r *http.Request)
 	})
 }
 
+// invalidateSharesForPlugin drops every cached body rendered from one plugin's
+// store. A mutating management call (save/delete/import/migrate) can change
+// what any of its records render to, and the content hash cannot see it — the
+// record, not the content, is what changed — so the edit path invalidates here.
+func (s *Server) invalidateSharesForPlugin(pluginID string) {
+	for _, share := range s.store.SubscriptionShares() {
+		if share.Source.Kind == model.ShareSourcePlugin && share.Source.PluginID == pluginID {
+			s.subscriptionCache.InvalidateShare(share.ID)
+		}
+	}
+}
+
+// invalidateSharesForSource drops cached bodies for shares sourcing one record.
+// The refresh path calls it when a fetch returns different bytes than the
+// stored snapshot.
+func (s *Server) invalidateSharesForSource(pluginID, subscriptionID string) {
+	for _, share := range s.store.SubscriptionShares() {
+		if share.Source.Kind == model.ShareSourcePlugin &&
+			share.Source.PluginID == pluginID && share.Source.SubscriptionID == subscriptionID {
+			s.subscriptionCache.InvalidateShare(share.ID)
+		}
+	}
+}
+
 // renderShare asks the share's source for content. It never shows the source the
 // token and never lets it influence the response beyond the bytes and a content
 // type.

internal/server/server_subscription_share_api_test.go

@@ -88,7 +88,7 @@ func TestRotateInvalidatesTheCachedBody(t *testing.T) {
 	s, _ := newShareTestServer(t)
 	share := mustCreateShare(t, s)
 	key := subscriptionCacheKey{ShareID: share.ID, Format: "base64", UAClass: "surge"}
-	s.subscriptionCache.Put(key, []byte("stale"), "text/plain", "", s.now())
+	s.subscriptionCache.Put(key, []byte("stale"), "text/plain", "", "", s.now())
 
 	rec := httptest.NewRecorder()
 	s.rotateSubscriptionShare(rec, share, principal{})

internal/server/subscription_cache.go

@@ -22,8 +22,14 @@ type subscriptionCacheEntry struct {
 	// userinfo is the provider's traffic header. It travels with the body rather
 	// than in a parallel map so a cache hit can never serve one client's body
 	// with another's remaining-quota figures.
-	userinfo  string
-	expiresAt time.Time
+	userinfo string
+	// contentHash is the render input's digest. On expiry the serve path
+	// re-fetches the (cheap) content and compares hashes: unchanged means the
+	// body is still exact and the entry is extended instead of re-rendered,
+	// which is what keeps a client poll from booting the plugin's JavaScript
+	// engine on every cycle.
+	contentHash string
+	expiresAt   time.Time
 }
 
 // subscriptionCache keeps rendered subscription bodies for a short time so a
@@ -65,17 +71,44 @@ func (c *subscriptionCache) Get(key subscriptionCacheKey, now time.Time) ([]byte
 	}
 	entry := el.Value.(*subscriptionCacheEntry)
 	if !now.Before(entry.expiresAt) {
-		c.removeElement(el)
+		// Expired is not deleted: the revalidation path may still extend this
+		// entry unchanged or serve it as the last good body. The LRU bound and
+		// the next Put are what reclaim it.
 		return nil, "", "", false
 	}
 	c.order.MoveToFront(el)
 	return entry.body, entry.contentType, entry.userinfo, true
 }
 
+// GetStale returns a copy of the entry whether or not it has expired. The
+// revalidation path uses it to extend an unchanged body — or to serve the last
+// good body when the source is unreachable — instead of dropping the client to
+// an error.
+func (c *subscriptionCache) GetStale(key subscriptionCacheKey) (subscriptionCacheEntry, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	el, ok := c.entries[key]
+	if !ok {
+		return subscriptionCacheEntry{}, false
+	}
+	return *el.Value.(*subscriptionCacheEntry), true
+}
+
+// Extend re-stamps an entry the serve path has revalidated against the current
+// content, so a subscription whose bytes did not move is never re-rendered.
+func (c *subscriptionCache) Extend(key subscriptionCacheKey, now time.Time) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if el, ok := c.entries[key]; ok {
+		el.Value.(*subscriptionCacheEntry).expiresAt = now.Add(c.ttl)
+		c.order.MoveToFront(el)
+	}
+}
+
 // Put ignores an empty body. The endpoint refuses to serve one, so letting it
 // into the cache would create a path back to the exact response that makes a
 // client delete every node it had.
-func (c *subscriptionCache) Put(key subscriptionCacheKey, body []byte, contentType, userinfo string, now time.Time) {
+func (c *subscriptionCache) Put(key subscriptionCacheKey, body []byte, contentType, userinfo, contentHash string, now time.Time) {
 	if len(body) == 0 {
 		return
 	}
@@ -89,6 +122,7 @@ func (c *subscriptionCache) Put(key subscriptionCacheKey, body []byte, contentTy
 		body:        body,
 		contentType: contentType,
 		userinfo:    userinfo,
+		contentHash: contentHash,
 		expiresAt:   now.Add(c.ttl),
 	})
 	c.entries[key] = el

internal/server/subscription_cache_test.go

@@ -13,7 +13,7 @@ func TestSubscriptionCacheServesFreshAndExpires(t *testing.T) {
 	base := time.Unix(1700000000, 0).UTC()
 	c := newSubscriptionCache(8, time.Minute)
 
-	c.Put(shareCacheKey("a"), []byte("body-a"), "text/plain", "", base)
+	c.Put(shareCacheKey("a"), []byte("body-a"), "text/plain", "", "", base)
 
 	body, ct, _, ok := c.Get(shareCacheKey("a"), base.Add(30*time.Second))
 	if !ok || string(body) != "body-a" || ct != "text/plain" {
@@ -28,9 +28,9 @@ func TestSubscriptionCacheIsBounded(t *testing.T) {
 	base := time.Unix(1700000000, 0).UTC()
 	c := newSubscriptionCache(2, time.Minute)
 
-	c.Put(shareCacheKey("x"), []byte("x"), "text/plain", "", base)
-	c.Put(shareCacheKey("y"), []byte("y"), "text/plain", "", base)
-	c.Put(shareCacheKey("z"), []byte("z"), "text/plain", "", base)
+	c.Put(shareCacheKey("x"), []byte("x"), "text/plain", "", "", base)
+	c.Put(shareCacheKey("y"), []byte("y"), "text/plain", "", "", base)
+	c.Put(shareCacheKey("z"), []byte("z"), "text/plain", "", "", base)
 
 	if c.Len() > 2 {
 		t.Fatalf("cache holds %d entries, cap is 2", c.Len())
@@ -50,11 +50,11 @@ func TestSubscriptionCacheNeverStoresEmptyBodies(t *testing.T) {
 	base := time.Unix(1700000000, 0).UTC()
 	c := newSubscriptionCache(4, time.Minute)
 
-	c.Put(shareCacheKey("a"), nil, "text/plain", "", base)
+	c.Put(shareCacheKey("a"), nil, "text/plain", "", "", base)
 	if _, _, _, ok := c.Get(shareCacheKey("a"), base); ok {
 		t.Fatal("a nil body was cached")
 	}
-	c.Put(shareCacheKey("a"), []byte{}, "text/plain", "", base)
+	c.Put(shareCacheKey("a"), []byte{}, "text/plain", "", "", base)
 	if _, _, _, ok := c.Get(shareCacheKey("a"), base); ok {
 		t.Fatal("an empty body was cached")
 	}
@@ -66,9 +66,9 @@ func TestSubscriptionCacheKeysOnFormatAndUAClass(t *testing.T) {
 	base := time.Unix(1700000000, 0).UTC()
 	c := newSubscriptionCache(8, time.Minute)
 
-	c.Put(subscriptionCacheKey{ShareID: "a", Format: "base64", UAClass: "surge"}, []byte("b64-surge"), "text/plain", "", base)
-	c.Put(subscriptionCacheKey{ShareID: "a", Format: "plain", UAClass: "surge"}, []byte("plain-surge"), "text/plain", "", base)
-	c.Put(subscriptionCacheKey{ShareID: "a", Format: "base64", UAClass: "loon"}, []byte("b64-loon"), "text/plain", "", base)
+	c.Put(subscriptionCacheKey{ShareID: "a", Format: "base64", UAClass: "surge"}, []byte("b64-surge"), "text/plain", "", "", base)
+	c.Put(subscriptionCacheKey{ShareID: "a", Format: "plain", UAClass: "surge"}, []byte("plain-surge"), "text/plain", "", "", base)
+	c.Put(subscriptionCacheKey{ShareID: "a", Format: "base64", UAClass: "loon"}, []byte("b64-loon"), "text/plain", "", "", base)
 
 	for _, tc := range []struct {
 		key  subscriptionCacheKey
@@ -88,9 +88,9 @@ func TestSubscriptionCacheKeysOnFormatAndUAClass(t *testing.T) {
 func TestSubscriptionCacheInvalidateShareDropsEveryFormat(t *testing.T) {
 	base := time.Unix(1700000000, 0).UTC()
 	c := newSubscriptionCache(8, time.Minute)
-	c.Put(subscriptionCacheKey{ShareID: "a", Format: "base64", UAClass: "surge"}, []byte("x"), "text/plain", "", base)
-	c.Put(subscriptionCacheKey{ShareID: "a", Format: "plain", UAClass: "loon"}, []byte("y"), "text/plain", "", base)
-	c.Put(subscriptionCacheKey{ShareID: "b", Format: "base64", UAClass: "surge"}, []byte("z"), "text/plain", "", base)
+	c.Put(subscriptionCacheKey{ShareID: "a", Format: "base64", UAClass: "surge"}, []byte("x"), "text/plain", "", "", base)
+	c.Put(subscriptionCacheKey{ShareID: "a", Format: "plain", UAClass: "loon"}, []byte("y"), "text/plain", "", "", base)
+	c.Put(subscriptionCacheKey{ShareID: "b", Format: "base64", UAClass: "surge"}, []byte("z"), "text/plain", "", "", base)
 
 	c.InvalidateShare("a")
 
@@ -111,8 +111,8 @@ func TestSubscriptionCacheInvalidateShareDropsEveryFormat(t *testing.T) {
 func TestSubscriptionCacheCarriesUserinfoWithTheBody(t *testing.T) {
 	base := time.Unix(1700000000, 0).UTC()
 	c := newSubscriptionCache(8, time.Minute)
-	c.Put(shareCacheKey("a"), []byte("a"), "text/plain", "upload=1; download=2; total=3", base)
-	c.Put(shareCacheKey("b"), []byte("b"), "text/plain", "upload=9", base)
+	c.Put(shareCacheKey("a"), []byte("a"), "text/plain", "upload=1; download=2; total=3", "", base)
+	c.Put(shareCacheKey("b"), []byte("b"), "text/plain", "upload=9", "", base)
 
 	_, _, ua, ok := c.Get(shareCacheKey("a"), base)
 	if !ok || ua != "upload=1; download=2; total=3" {
@@ -123,3 +123,43 @@ func TestSubscriptionCacheCarriesUserinfoWithTheBody(t *testing.T) {
 		t.Fatalf("userinfo for b = %q (ok=%v)", ub, ok)
 	}
 }
+
+func TestSubscriptionCacheRevalidationExtendsUnchangedBody(t *testing.T) {
+	base := time.Unix(1700000000, 0).UTC()
+	c := newSubscriptionCache(8, time.Minute)
+	key := shareCacheKey("a")
+	c.Put(key, []byte("body-a"), "text/plain", "ui", "hash-1", base)
+
+	// Past expiry the plain Get misses, but the stale entry is still readable
+	// for the revalidation decision.
+	if _, _, _, ok := c.Get(key, base.Add(2*time.Minute)); ok {
+		t.Fatal("expired entry served without revalidation")
+	}
+	stale, ok := c.GetStale(key)
+	if !ok || string(stale.body) != "body-a" || stale.contentHash != "hash-1" {
+		t.Fatalf("stale entry = %q %q %v", stale.body, stale.contentHash, ok)
+	}
+
+	// The serve path's "hash still matches" branch: extend, and the entry
+	// serves again for a full TTL from the extension.
+	c.Extend(key, base.Add(2*time.Minute))
+	body, _, ui, ok := c.Get(key, base.Add(2*time.Minute+30*time.Second))
+	if !ok || string(body) != "body-a" || ui != "ui" {
+		t.Fatalf("extended entry not served: %q %q %v", body, ui, ok)
+	}
+}
+
+func TestSubscriptionCacheHashChangeForcesReplace(t *testing.T) {
+	base := time.Unix(1700000000, 0).UTC()
+	c := newSubscriptionCache(8, time.Minute)
+	key := shareCacheKey("a")
+	c.Put(key, []byte("old"), "text/plain", "", "hash-1", base)
+
+	// The "hash moved" branch re-renders and Puts under the new hash; a later
+	// revalidation against the old hash must not resurrect the old body.
+	c.Put(key, []byte("new"), "text/plain", "", "hash-2", base.Add(2*time.Minute))
+	stale, ok := c.GetStale(key)
+	if !ok || string(stale.body) != "new" || stale.contentHash != "hash-2" {
+		t.Fatalf("replaced entry = %q %q %v", stale.body, stale.contentHash, ok)
+	}
+}

internal/server/subscription_refresh.go

@@ -70,6 +70,12 @@ func (s *Server) snapshotFor(ctx context.Context, pluginID, subscriptionID strin
 	if err := s.store.UpsertSubscriptionSnapshot(fetched); err != nil {
 		return model.SubscriptionSnapshot{}, err
 	}
+	// The content moved: any rendered body cached for a share sourcing this
+	// record is now stale, no matter how much TTL it had left. Without this the
+	// revalidation cadence, not the content, would decide what clients get.
+	if has && existing.Raw != fetched.Raw {
+		s.invalidateSharesForSource(pluginID, subscriptionID)
+	}
 	s.recordAudit(model.AuditEvent{
 		ID: id.New("audit"), Action: auditActionSubscriptionFetch, Decision: "allow",
 		Metadata: map[string]string{

internal/server/subscription_share_invalidate_test.go

@@ -0,0 +1,59 @@
+package server
+
+import (
+	"testing"
+	"time"
+
+	"github.com/LatticeNet/lattice-sdk/model"
+)
+
+// The rendered-body cache must drop entries the moment their inputs move:
+// a provider refresh that returned different bytes (source invalidation), or
+// an operator edit through the plugin's management API (plugin invalidation).
+// TTL is only the revalidation cadence; change is the real invalidator.
+func TestShareCacheInvalidationBySourceAndPlugin(t *testing.T) {
+	s, st := newShareTestServer(t)
+	base := s.now()
+	mustUpsertShare(t, st, model.SubscriptionShare{
+		ID: "s1", Slug: "one", Token: "t1", Enabled: true,
+		Source: model.ShareSource{Kind: model.ShareSourcePlugin, PluginID: "latticenet.sub-store", SubscriptionID: "sub-a"},
+	})
+	mustUpsertShare(t, st, model.SubscriptionShare{
+		ID: "s2", Slug: "two", Token: "t2", Enabled: true,
+		Source: model.ShareSource{Kind: model.ShareSourcePlugin, PluginID: "latticenet.sub-store", SubscriptionID: "sub-b"},
+	})
+	mustUpsertShare(t, st, model.SubscriptionShare{
+		ID: "s3", Slug: "three", Token: "t3", Enabled: true,
+		Source: model.ShareSource{Kind: model.ShareSourcePlugin, PluginID: "latticenet.other", SubscriptionID: "sub-a"},
+	})
+
+	put := func(id string) {
+		s.subscriptionCache.Put(subscriptionCacheKey{ShareID: id, Format: "base64", UAClass: "surge"}, []byte("x"), "text/plain", "", "h", base)
+	}
+	put("s1")
+	put("s2")
+	put("s3")
+
+	// A content change on sub-a drops exactly the shares sourcing it.
+	s.invalidateSharesForSource("latticenet.sub-store", "sub-a")
+	if _, ok := s.subscriptionCache.GetStale(subscriptionCacheKey{ShareID: "s1", Format: "base64", UAClass: "surge"}); ok {
+		t.Fatal("s1 survived its source's content change")
+	}
+	for _, id := range []string{"s2", "s3"} {
+		if _, ok := s.subscriptionCache.GetStale(subscriptionCacheKey{ShareID: id, Format: "base64", UAClass: "surge"}); !ok {
+			t.Fatalf("%s was dropped by an unrelated source change", id)
+		}
+	}
+
+	// A store mutation through the plugin's management API drops every share
+	// sourcing that plugin, and no one else's.
+	s.invalidateSharesForPlugin("latticenet.sub-store")
+	if _, ok := s.subscriptionCache.GetStale(subscriptionCacheKey{ShareID: "s2", Format: "base64", UAClass: "surge"}); ok {
+		t.Fatal("s2 survived its plugin's store mutation")
+	}
+	if _, ok := s.subscriptionCache.GetStale(subscriptionCacheKey{ShareID: "s3", Format: "base64", UAClass: "surge"}); !ok {
+		t.Fatal("s3 was dropped by another plugin's mutation")
+	}
+}
+
+var _ = time.Minute // keep the import if the file grows a TTL case