perf: memoize the Lines read model (explicit invalidation + 60s TTL net)

internal/server/lines_cache.go

@@ -0,0 +1,73 @@
+package server
+
+import (
+	"sync"
+	"time"
+)
+
+// lines_cache.go — a cheap read-model cache for the unified Lines view
+// (design-15 follow-up). buildLineGroups walks every store collection and the
+// whole fleet inventory on every call, and lines.get used to redo that fleet
+// build just to linear-scan one line. The cache holds the last built groups
+// plus a line_hash_id index, and is invalidated EXPLICITLY on every state
+// change that feeds the read model: proxy inbound/user/profile writes, node
+// writes, sing-box inventory ingest, and vpn-core identity mutations. A 60s
+// TTL is the documented safety net for a missed edge path — operator-driven
+// mutations all invalidate explicitly, so the UI never reads stale after its
+// own actions.
+
+const lineReadModelTTL = 60 * time.Second
+
+type lineReadModelCache struct {
+	mu      sync.RWMutex
+	groups  []LineGroup
+	byHash  map[string]Line
+	builtAt time.Time
+	valid   bool
+}
+
+// lineReadModel returns the cached Lines view, rebuilding it after an
+// invalidation or when the TTL safety net expired. Callers must not mutate the
+// returned slices.
+func (s *Server) lineReadModel() ([]LineGroup, map[string]Line) {
+	now := s.now()
+	s.lineCache.mu.RLock()
+	if s.lineCache.valid && now.Sub(s.lineCache.builtAt) < lineReadModelTTL {
+		defer s.lineCache.mu.RUnlock()
+		return s.lineCache.groups, s.lineCache.byHash
+	}
+	s.lineCache.mu.RUnlock()
+	groups := s.buildLineGroups()
+	index := make(map[string]Line, 64)
+	for _, g := range groups {
+		for _, ln := range g.Lines {
+			if ln.LineHashID != "" {
+				index[ln.LineHashID] = ln
+			}
+		}
+	}
+	s.lineCache.mu.Lock()
+	s.lineCache.groups = groups
+	s.lineCache.byHash = index
+	s.lineCache.builtAt = now
+	s.lineCache.valid = true
+	s.lineCache.mu.Unlock()
+	return groups, index
+}
+
+// invalidateLineReadModel marks the Lines view stale. It is called on every
+// state change the view derives from (see file header).
+func (s *Server) invalidateLineReadModel() {
+	s.lineCache.mu.Lock()
+	s.lineCache.valid = false
+	s.lineCache.groups = nil
+	s.lineCache.byHash = nil
+	s.lineCache.mu.Unlock()
+}
+
+// lineFromReadModel resolves one line by hash without a fleet rebuild.
+func (s *Server) lineFromReadModel(lineHashID string) (Line, bool) {
+	_, index := s.lineReadModel()
+	ln, ok := index[lineHashID]
+	return ln, ok
+}

internal/server/lines_cache_test.go

@@ -0,0 +1,105 @@
+package server
+
+import (
+	"testing"
+	"time"
+
+	"github.com/LatticeNet/lattice-sdk/model"
+	"github.com/LatticeNet/lattice-server/internal/store"
+)
+
+// The read model is built once and served from cache until an invalidation;
+// line lookups go through the hash index, not a fleet rescan.
+func TestLineReadModelCacheLifecycle(t *testing.T) {
+	st, err := store.Open("")
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := newLinemetaTestServer(t, st)
+	seedLinemetaNodes(t, srv)
+
+	groups, index := srv.lineReadModel()
+	hub := findLine(t, groups, "node-a", "hub-a")
+	if index[hub.LineHashID].Tag != "hub-a" {
+		t.Fatalf("index: %+v", index[hub.LineHashID])
+	}
+
+	// Inventory changes are invisible until invalidation.
+	srv.singboxInvMu.Lock()
+	inv := srv.singboxInv["node-a"]
+	inv.Nodes = append(inv.Nodes, model.SingBoxNode{Name: "new-line", Protocol: "trojan", Port: "9999", Address: "203.0.113.5"})
+	srv.singboxInv["node-a"] = inv
+	srv.singboxInvMu.Unlock()
+	cached, _ := srv.lineReadModel()
+	for _, g := range cached {
+		for _, ln := range g.Lines {
+			if ln.Tag == "new-line" {
+				t.Fatal("cache must not rebuild on its own")
+			}
+		}
+	}
+	srv.invalidateLineReadModel()
+	fresh, freshIndex := srv.lineReadModel()
+	found := false
+	for _, g := range fresh {
+		for _, ln := range g.Lines {
+			if ln.Tag == "new-line" {
+				found = true
+				if freshIndex[ln.LineHashID].Tag != "new-line" {
+					t.Fatalf("index miss after rebuild: %+v", freshIndex)
+				}
+			}
+		}
+	}
+	if !found {
+		t.Fatal("new line missing after invalidation")
+	}
+
+	// lineFromReadModel resolves via the index.
+	if _, ok := srv.lineFromReadModel(hub.LineHashID); !ok {
+		t.Fatal("index lookup failed")
+	}
+
+	// The TTL safety net rebuilds even without an explicit invalidation.
+	srv.lineCache.mu.Lock()
+	srv.lineCache.builtAt = time.Now().Add(-2 * lineReadModelTTL)
+	srv.lineCache.mu.Unlock()
+	srv.singboxInvMu.Lock()
+	inv.Nodes = append(inv.Nodes, model.SingBoxNode{Name: "ttl-line", Protocol: "vless", Port: "9998", Address: "203.0.113.5"})
+	srv.singboxInv["node-a"] = inv
+	srv.singboxInvMu.Unlock()
+	afterTTL, _ := srv.lineReadModel()
+	seen := false
+	for _, g := range afterTTL {
+		for _, ln := range g.Lines {
+			if ln.Tag == "ttl-line" {
+				seen = true
+			}
+		}
+	}
+	if !seen {
+		t.Fatal("TTL expiry must force a rebuild")
+	}
+}
+
+// Invalidations wired into the mutation paths: an inventory ingest marks the
+// model stale (the discover handler does this before queueing any sync).
+func TestInventoryIngestInvalidates(t *testing.T) {
+	st, err := store.Open("")
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := newLinemetaTestServer(t, st)
+	seedLinemetaNodes(t, srv)
+	srv.lineReadModel()
+	if !srv.lineCache.valid {
+		t.Fatal("cache should be valid after first build")
+	}
+	srv.singboxInvMu.Lock()
+	srv.singboxInv["node-b"] = model.SingBoxInventory{NodeID: "node-b", Status: "ok"}
+	srv.singboxInvMu.Unlock()
+	srv.invalidateLineReadModel()
+	if srv.lineCache.valid {
+		t.Fatal("invalidation must mark the cache stale")
+	}
+}

internal/server/lineusers.go

@@ -137,7 +137,8 @@ func lineUserCredentialSHA(payload lineUserCredentialPayload) (string, error) {
 // lines take the whole-config render path (design-15 D6 deferred), so they are
 // rejected here with an explicit error rather than silently mis-routed.
 func (s *Server) resolveAdoptedLine(lineHashID string) (Line, error) {
-	for _, g := range s.buildLineGroups() {
+	groups, _ := s.lineReadModel()
+	for _, g := range groups {
 		for _, ln := range g.Lines {
 			if ln.LineHashID != lineHashID {
 				continue

internal/server/profiles.go

@@ -249,6 +249,7 @@ func (s *Server) vpnCoreProfilesRPC(ctx context.Context, method string, request
 		if err := s.store.UpsertNode(node); err != nil {
 			return nil, err
 		}
+		s.invalidateLineReadModel()
 		settings, err := s.vpnCoreProfileSettings(req.NodeID)
 		if err != nil {
 			return nil, err

internal/server/server.go

@@ -161,6 +161,9 @@ type Server struct {
 	// after committed vpn-core mutations (design-15 §7). Nil-safe: trigger and
 	// fire paths both tolerate it.
 	subStoreSync *subStoreSyncState
+	// lineCache memoizes the unified Lines read model until an explicit
+	// invalidation (lines_cache.go).
+	lineCache lineReadModelCache
 	// linemetaSyncFP tracks the last-queued discovery fingerprint per node so a
 	// sidecar sync is queued only when the discovered line set actually changed.
 	linemetaSyncMu sync.Mutex
@@ -1993,6 +1996,7 @@ func (s *Server) ensureNodeIdentityUUID(nodeID string) (string, error) {
 	if err := s.store.UpsertNode(n); err != nil {
 		return "", err
 	}
+	s.invalidateLineReadModel()
 	return n.LatticeIdentityUUID, nil
 }
 
@@ -2243,6 +2247,7 @@ func (s *Server) handleEnrollNode(w http.ResponseWriter, r *http.Request, p prin
 		writeError(w, http.StatusInternalServerError, err)
 		return
 	}
+	s.invalidateLineReadModel()
 	// Append the new node into each requested group's explicit Members — the same
 	// canonical membership path handleGroupMembers uses. Idempotent: a node that
 	// is already a member is left untouched.

internal/server/server_node_delete.go

@@ -141,6 +141,7 @@ func (s *Server) handleDeleteNode(w http.ResponseWriter, r *http.Request, p prin
 		writeError(w, http.StatusInternalServerError, err)
 		return
 	}
+	s.invalidateLineReadModel()
 	if !ok {
 		writeError(w, http.StatusNotFound, errors.New("node not found"))
 		return

internal/server/server_proxy.go

@@ -378,6 +378,7 @@ func (s *Server) handleProxyInbounds(w http.ResponseWriter, r *http.Request, p p
 			writeError(w, http.StatusInternalServerError, err)
 			return
 		}
+		s.invalidateLineReadModel()
 		if stored, ok := s.store.ProxyInbound(inbound.ID); ok {
 			inbound = stored
 		}
@@ -430,6 +431,7 @@ func (s *Server) handleDeleteProxyInbound(w http.ResponseWriter, r *http.Request
 		writeError(w, http.StatusInternalServerError, err)
 		return
 	}
+	s.invalidateLineReadModel()
 	s.recordPrincipalAudit(p, model.AuditEvent{
 		ID:       id.New("audit"),
 		Action:   "proxy.inbound.delete",
@@ -472,6 +474,7 @@ func (s *Server) handleProxyUsers(w http.ResponseWriter, r *http.Request, p prin
 			writeError(w, http.StatusInternalServerError, err)
 			return
 		}
+		s.invalidateLineReadModel()
 		if stored, ok := s.store.ProxyUser(user.ID); ok {
 			user = stored
 		}
@@ -510,6 +513,7 @@ func (s *Server) handleDeleteProxyUser(w http.ResponseWriter, r *http.Request, p
 		writeError(w, http.StatusInternalServerError, err)
 		return
 	}
+	s.invalidateLineReadModel()
 	s.recordPrincipalAudit(p, model.AuditEvent{
 		ID:       id.New("audit"),
 		Action:   "proxy.user.delete",
@@ -687,6 +691,7 @@ func (s *Server) handleProxyProfiles(w http.ResponseWriter, r *http.Request, p p
 			writeError(w, http.StatusInternalServerError, err)
 			return
 		}
+		s.invalidateLineReadModel()
 		if stored, ok := s.store.ProxyNodeProfile(profile.NodeID); ok {
 			profile = stored
 		}
@@ -726,6 +731,7 @@ func (s *Server) handleDeleteProxyProfile(w http.ResponseWriter, r *http.Request
 		writeError(w, http.StatusInternalServerError, err)
 		return
 	}
+	s.invalidateLineReadModel()
 	s.recordPrincipalAudit(p, model.AuditEvent{
 		ID:       id.New("audit"),
 		NodeID:   req.NodeID,
@@ -1117,6 +1123,7 @@ func (s *Server) handleProxyCoreTaskResult(r *http.Request, approval model.Appro
 		if err := s.store.UpsertProxyNodeProfile(profile); err != nil {
 			return fmt.Errorf("mark proxycore profile applied: %w", err)
 		}
+		s.invalidateLineReadModel()
 		s.recordRequestAudit(r, model.AuditEvent{
 			ID:       id.New("audit"),
 			NodeID:   approval.NodeID,
@@ -1134,6 +1141,7 @@ func (s *Server) handleProxyCoreTaskResult(r *http.Request, approval model.Appro
 	if err := s.store.UpsertProxyNodeProfile(profile); err != nil {
 		return fmt.Errorf("mark proxycore apply failed: %w", err)
 	}
+	s.invalidateLineReadModel()
 	if err := s.rejectApprovalWithReason(approval, reason); err != nil {
 		return fmt.Errorf("mark proxycore approval rejected: %w", err)
 	}
@@ -1496,7 +1504,8 @@ func (s *Server) sanitizeProxyUsageLineUserBytes(input map[string]map[string]int
 		return nil, nil, 0, errors.New("line_user_bytes has too many lines")
 	}
 	knownLines := map[string]bool{}
-	for _, group := range s.buildLineGroups() {
+	groups, _ := s.lineReadModel()
+	for _, group := range groups {
 		for _, line := range group.Lines {
 			if line.LineHashID != "" {
 				knownLines[line.LineHashID] = true

internal/server/server_proxy_notify.go

@@ -32,6 +32,7 @@ func (s *Server) evaluateProxyUserNotifications(now time.Time, onlyID string) ([
 	users := s.store.ProxyUsers()
 	fired := []proxyUserNotificationFire{}
 	found := onlyID == ""
+	changed := false
 	for _, user := range users {
 		if onlyID != "" && user.ID != onlyID {
 			continue
@@ -45,15 +46,20 @@ func (s *Server) evaluateProxyUserNotifications(now time.Time, onlyID string) ([
 				if err := s.store.UpsertProxyUser(updated); err != nil {
 					return nil, err
 				}
+				changed = true
 			}
 			continue
 		}
 		if err := s.store.UpsertProxyUser(updated); err != nil {
 			return nil, err
 		}
+		changed = true
 		s.emitProxyUserNotifications(alerts)
 		fired = append(fired, alerts...)
 	}
+	if changed {
+		s.invalidateLineReadModel()
+	}
 	if !found {
 		return nil, fmt.Errorf("proxy user not found")
 	}

internal/server/server_singbox_discover.go

@@ -60,6 +60,7 @@ func (s *Server) handleAgentSingBoxInventory(w http.ResponseWriter, r *http.Requ
 	}
 	s.singboxInv[req.NodeID] = inv
 	s.singboxInvMu.Unlock()
+	s.invalidateLineReadModel()
 
 	// design-15 D2: a changed line set queues a sidecar sync (pending approval —
 	// discovery still never mutates the node by itself).

internal/server/server_vpncore.go

@@ -69,7 +69,7 @@ func (s *Server) registerVPNCoreRPC() {
 func (s *Server) vpnCoreLinesRPC(ctx context.Context, method string, request []byte) ([]byte, error) {
 	switch method {
 	case "list":
-		groups := s.buildLineGroups()
+		groups, _ := s.lineReadModel()
 		count := 0
 		for _, g := range groups {
 			count += len(g.Lines)
@@ -96,14 +96,10 @@ func (s *Server) vpnCoreLinesRPC(ctx context.Context, method string, request []b
 		if strings.TrimSpace(req.LineHashID) == "" {
 			return nil, fmt.Errorf("vpn-core/lines get: line_hash_id required")
 		}
-		for _, g := range s.buildLineGroups() {
-			for _, ln := range g.Lines {
-				if ln.LineHashID == req.LineHashID {
-					return json.Marshal(struct {
-						Line Line `json:"line"`
-					}{Line: ln})
-				}
-			}
+		if ln, ok := s.lineFromReadModel(req.LineHashID); ok {
+			return json.Marshal(struct {
+				Line Line `json:"line"`
+			}{Line: ln})
 		}
 		return nil, fmt.Errorf("vpn-core/lines get: line %q not found", req.LineHashID)
 	default:

internal/server/vpnusers.go

@@ -251,6 +251,7 @@ func (s *Server) vpnCoreUsersAdminRPC(ctx context.Context, method string, reques
 		switch method {
 		case "create", "update", "delete", "bind", "unbind", "rotate":
 			s.triggerVPNCoreMutation()
+			s.invalidateLineReadModel()
 		}
 	}
 	return out, err
@@ -501,14 +502,8 @@ func (s *Server) vpnUserEmailInUse(email, exceptID string) bool {
 
 // lineExists reports whether a line_hash_id is currently present on any node.
 func (s *Server) lineExists(lineHash string) bool {
-	for _, g := range s.buildLineGroups() {
-		for _, ln := range g.Lines {
-			if ln.LineHashID == lineHash {
-				return true
-			}
-		}
-	}
-	return false
+	_, ok := s.lineFromReadModel(lineHash)
+	return ok
 }
 
 // normalizeCredentials validates protocols and secret material, auto-generating a