From 68670a9a2f22e090ea9a700b301c893fc3582b00 Mon Sep 17 00:00:00 2001 From: lr00rl Date: Tue, 21 Jul 2026 21:50:51 -0700 Subject: [PATCH] Memoize the Lines read model with explicit invalidation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unified Lines view used to rebuild fleet-wide on every call — lines.get included a full rebuild plus a linear scan for a single hash. A small cache now holds the built groups and a line_hash_id index, rebuilt only after explicit invalidation: inventory ingest, proxy inbound/user/profile writes, node writes, vpn-core identity mutations, and apply outcomes. 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. --- internal/server/lines_cache.go | 73 ++++++++++++++ internal/server/lines_cache_test.go | 105 +++++++++++++++++++++ internal/server/lineusers.go | 3 +- internal/server/profiles.go | 1 + internal/server/server.go | 5 + internal/server/server_node_delete.go | 1 + internal/server/server_proxy.go | 11 ++- internal/server/server_proxy_notify.go | 6 ++ internal/server/server_singbox_discover.go | 1 + internal/server/server_vpncore.go | 14 +-- internal/server/vpnusers.go | 11 +-- 11 files changed, 212 insertions(+), 19 deletions(-) create mode 100644 internal/server/lines_cache.go create mode 100644 internal/server/lines_cache_test.go diff --git a/internal/server/lines_cache.go b/internal/server/lines_cache.go new file mode 100644 index 0000000..5173ab3 --- /dev/null +++ b/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 +} diff --git a/internal/server/lines_cache_test.go b/internal/server/lines_cache_test.go new file mode 100644 index 0000000..ccef766 --- /dev/null +++ b/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") + } +} diff --git a/internal/server/lineusers.go b/internal/server/lineusers.go index 5c07f41..530b2c5 100644 --- a/internal/server/lineusers.go +++ b/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 diff --git a/internal/server/profiles.go b/internal/server/profiles.go index 0091855..cb50a98 100644 --- a/internal/server/profiles.go +++ b/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 diff --git a/internal/server/server.go b/internal/server/server.go index 2d4418d..a0a97dd 100644 --- a/internal/server/server.go +++ b/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. diff --git a/internal/server/server_node_delete.go b/internal/server/server_node_delete.go index e4d720c..8ef54cc 100644 --- a/internal/server/server_node_delete.go +++ b/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 diff --git a/internal/server/server_proxy.go b/internal/server/server_proxy.go index 9f1849e..faecdfd 100644 --- a/internal/server/server_proxy.go +++ b/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 diff --git a/internal/server/server_proxy_notify.go b/internal/server/server_proxy_notify.go index c8e2ae1..48bebd6 100644 --- a/internal/server/server_proxy_notify.go +++ b/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") } diff --git a/internal/server/server_singbox_discover.go b/internal/server/server_singbox_discover.go index 4da9fa7..6592f30 100644 --- a/internal/server/server_singbox_discover.go +++ b/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). diff --git a/internal/server/server_vpncore.go b/internal/server/server_vpncore.go index 12d3e8a..b0a5f95 100644 --- a/internal/server/server_vpncore.go +++ b/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: diff --git a/internal/server/vpnusers.go b/internal/server/vpnusers.go index 394028e..96d9e40 100644 --- a/internal/server/vpnusers.go +++ b/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