package distsync import ( "context" "errors" "fmt" "reflect" "time" "github.com/mudler/LocalAI/core/services/failover" "github.com/mudler/LocalAI/core/services/messaging" "github.com/mudler/LocalAI/core/services/syncstate" "github.com/mudler/xlog" ) // compile-time assertions. var ( _ syncstate.Store[string, PinRecord] = (*PinStore)(nil) _ failover.StateSync = (*Sync)(nil) ) // pinReconcileInterval is how often the pins map re-reads the DB. const pinReconcileInterval = 30 * time.Second // Sync is a failover.StateSync backed by three syncstate.SyncedMaps: // // - failover.pins keeps the durable source of truth: Store-backed (when a // PinStore is given) so a late joiner hydrates every pin from the DB on // Start, not just from whatever peers happen to broadcast afterwards, // and re-reads it every pinReconcileInterval to repair a missed delta. // - failover.targets and failover.chains are ephemeral live-health state, // NATS-only with no Store and no Reconcile: with neither set, a Reconcile // tick's hydrate is a no-op (nothing durable to pull from), so it could // never help a late joiner catch up anyway. A late joiner instead catches // up from the leader's periodic Republish (see failover.Manager.Republish). type Sync struct { pins *syncstate.SyncedMap[string, PinRecord] targets *syncstate.SyncedMap[string, failover.TargetSnapshot] chains *syncstate.SyncedMap[string, failover.ChainSnapshot] } // New builds and starts the three maps, then attaches the result to m via // SetStateSync so any already-durable pins hydrate onto m immediately. func New(ctx context.Context, nats messaging.MessagingClient, pins syncstate.Store[string, PinRecord], m *failover.Manager) (*Sync, error) { s := &Sync{} // pins is already typed as the Store interface (the brief fixes this // signature), so a caller holding a nil *PinStore (e.g. standalone mode, // no DB configured) and passing it straight through boxes it into a // non-nil interface wrapping a nil pointer - "pins != nil" alone would // not catch that, and the SyncedMap would then try to hydrate/write // through a nil *gorm.DB. isNilStore catches both that and a literal nil // argument, the same defense finetune/service.go gets for free by taking // a concrete *distributed.FineTuneStore and nil-checking before boxing it. var pinStore syncstate.Store[string, PinRecord] if !isNilStore(pins) { pinStore = pins } // A delta dropped without a reconnect would leave this map stale until // the next reconnect; re-reading the DB repairs it, and the manager's // periodic ReconcilePins carries the repair into the chains. var reconcile time.Duration if pinStore != nil { reconcile = pinReconcileInterval } s.pins = syncstate.New(syncstate.Config[string, PinRecord]{ Name: "failover.pins", Key: func(p PinRecord) string { return p.Chain }, Nats: nats, Store: pinStore, Reconcile: reconcile, OnApply: func(op string, chain string, v PinRecord) { if op == "delete" { m.ApplyPin(chain, "") return } m.ApplyPin(chain, v.Target) }, }) if err := s.pins.Start(ctx); err != nil { return nil, fmt.Errorf("distsync: starting pins map: %w", err) } s.targets = syncstate.New(syncstate.Config[string, failover.TargetSnapshot]{ Name: "failover.targets", Key: func(t failover.TargetSnapshot) string { return t.Target }, Nats: nats, OnApply: func(_ string, _ string, v failover.TargetSnapshot) { m.ApplyTarget(v) }, }) if err := s.targets.Start(ctx); err != nil { _ = s.pins.Close() return nil, fmt.Errorf("distsync: starting targets map: %w", err) } s.chains = syncstate.New(syncstate.Config[string, failover.ChainSnapshot]{ Name: "failover.chains", Key: func(c failover.ChainSnapshot) string { return c.Chain }, Nats: nats, OnApply: func(_ string, _ string, v failover.ChainSnapshot) { m.ApplyChain(v) }, }) if err := s.chains.Start(ctx); err != nil { _ = s.targets.Close() _ = s.pins.Close() return nil, fmt.Errorf("distsync: starting chains map: %w", err) } m.SetStateSync(s) // The pins map re-hydrates from the DB on reconnect without OnApply, so // hand the manager the result. Registered after the map's own callback, // which runs first. if r, ok := nats.(interface{ OnReconnect(func()) }); ok { r.OnReconnect(m.ReconcilePins) } return s, nil } // isNilStore reports whether pins is nil - either a literal nil argument, or // the classic Go footgun of a non-nil interface value wrapping a nil // pointer (e.g. a nil *PinStore passed in directly). Both must disable the // durable Store the same way, so syncstate hydrates from nothing rather than // panicking on a nil *gorm.DB the first time it dereferences it. func isNilStore(pins syncstate.Store[string, PinRecord]) bool { if pins == nil { return true } v := reflect.ValueOf(pins) return v.Kind() == reflect.Ptr && v.IsNil() } // Close releases all three maps' subscriptions and background workers. func (s *Sync) Close() error { return errors.Join(s.chains.Close(), s.targets.Close(), s.pins.Close()) } // PublishTarget shares a target health transition. StateSync's methods // return no error to the manager, and a publish must never block it (the // manager calls this outside its lock precisely so a synchronous NATS echo // is safe) - so a failure here is logged and dropped; a missed publish // self-heals on the leader's next Republish. func (s *Sync) PublishTarget(t failover.TargetSnapshot) { if err := s.targets.Set(context.Background(), t); err != nil { xlog.Warn("distsync: publishing target state failed", "target", t.Target, "error", err) } } // PublishChain shares the leader's decision for a chain. func (s *Sync) PublishChain(c failover.ChainSnapshot) { if err := s.chains.Set(context.Background(), c); err != nil { xlog.Warn("distsync: publishing chain state failed", "chain", c.Chain, "error", err) } } // SetPin durably persists and broadcasts a pin. func (s *Sync) SetPin(chain, target string) error { return s.pins.Set(context.Background(), PinRecord{Chain: chain, Target: target, UpdatedAt: time.Now()}) } // ClearPin durably removes and broadcasts a pin's removal. func (s *Sync) ClearPin(chain string) error { return s.pins.Delete(context.Background(), chain) } // Pins returns every known pin (chain -> target), for Manager.SetStateSync's // hydrate-on-attach. func (s *Sync) Pins() map[string]string { out := make(map[string]string) for chain, rec := range s.pins.Snapshot() { out[chain] = rec.Target } return out }