wehub-resource-sync
92 lines
3.7 KiB
Go
92 lines
3.7 KiB
Go
package prefixcache
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/mudler/LocalAI/core/services/messaging"
|
|
"github.com/mudler/xlog"
|
|
)
|
|
|
|
// publisher is the minimal slice of messaging.Client that Sync needs.
|
|
type publisher interface {
|
|
Publish(subject string, v any) error
|
|
}
|
|
|
|
// Sync wraps an Index, broadcasting new/extended observations to peers and
|
|
// applying peers' broadcasts. It is the cross-frontend coherence layer.
|
|
type Sync struct {
|
|
idx Provider
|
|
pub publisher
|
|
}
|
|
|
|
func NewSync(idx Provider, pub publisher) *Sync { return &Sync{idx: idx, pub: pub} }
|
|
|
|
// Observe records locally and, if new/extended, broadcasts to peers. It returns
|
|
// whether the local index treated the assignment as new or extended, so Sync
|
|
// satisfies prefixcache.Provider.
|
|
func (s *Sync) Observe(model string, chain []uint64, key ReplicaKey, now time.Time) bool {
|
|
changed := s.idx.Observe(model, chain, key, now)
|
|
if changed && s.pub != nil {
|
|
ev := messaging.PrefixCacheObserveEvent{Model: model, Chain: chain, NodeID: key.NodeID, Replica: key.Replica}
|
|
if err := s.pub.Publish(messaging.SubjectPrefixCacheObserve, ev); err != nil {
|
|
xlog.Debug("prefixcache: observe publish failed", "error", err)
|
|
}
|
|
}
|
|
return changed
|
|
}
|
|
|
|
// Invalidate drops the local entry for one replica and broadcasts to peers. The
|
|
// local drop is a no-op for models that were never cached (Index.Invalidate does
|
|
// not intern a tree). The broadcast is UNCONDITIONAL (when a publisher is
|
|
// configured): the registry chokepoint fires for every replica removal, and a
|
|
// peer frontend may hold a stale entry for the model even when THIS frontend
|
|
// never cached it, so gating the broadcast on local-tree existence would drop
|
|
// cross-frontend invalidations and leave peers routing to a removed replica
|
|
// until their TTL.
|
|
func (s *Sync) Invalidate(model string, key ReplicaKey) {
|
|
s.idx.Invalidate(model, key)
|
|
if s.pub != nil {
|
|
ev := messaging.PrefixCacheInvalidateEvent{Model: model, NodeID: key.NodeID, Replica: key.Replica}
|
|
if err := s.pub.Publish(messaging.SubjectPrefixCacheInvalidate, ev); err != nil {
|
|
xlog.Debug("prefixcache: invalidate publish failed", "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// InvalidateNode drops the local entries for ALL replicas of node and broadcasts
|
|
// to peers. Like Invalidate the broadcast is unconditional for cross-frontend
|
|
// coherence. A negative Replica on the wire means "all replicas of the node".
|
|
func (s *Sync) InvalidateNode(model, node string) {
|
|
s.idx.InvalidateNode(model, node)
|
|
if s.pub != nil {
|
|
ev := messaging.PrefixCacheInvalidateEvent{Model: model, NodeID: node, Replica: -1}
|
|
if err := s.pub.Publish(messaging.SubjectPrefixCacheInvalidate, ev); err != nil {
|
|
xlog.Debug("prefixcache: invalidate-node publish failed", "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// ApplyObserve applies a peer observe event locally (no re-broadcast).
|
|
func (s *Sync) ApplyObserve(ev messaging.PrefixCacheObserveEvent, now time.Time) {
|
|
s.idx.Observe(ev.Model, ev.Chain, ReplicaKey{NodeID: ev.NodeID, Replica: ev.Replica}, now)
|
|
}
|
|
|
|
// ApplyInvalidate applies a peer invalidate event locally (no re-broadcast). A
|
|
// negative Replica targets all replicas of the node.
|
|
func (s *Sync) ApplyInvalidate(ev messaging.PrefixCacheInvalidateEvent) {
|
|
if ev.Replica < 0 {
|
|
s.idx.InvalidateNode(ev.Model, ev.NodeID)
|
|
return
|
|
}
|
|
s.idx.Invalidate(ev.Model, ReplicaKey{NodeID: ev.NodeID, Replica: ev.Replica})
|
|
}
|
|
|
|
// Decide delegates to the wrapped index.
|
|
func (s *Sync) Decide(model string, chain []uint64, candidates []ReplicaKey, now time.Time) PrefixDecision {
|
|
return s.idx.Decide(model, chain, candidates, now)
|
|
}
|
|
|
|
// Evict delegates eviction of expired entries to the wrapped index. It does not
|
|
// broadcast: each frontend evicts its own copy on its own TTL clock.
|
|
func (s *Sync) Evict(now time.Time) { s.idx.Evict(now) }
|