mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-21 21:54:52 -04:00
Ettore Di Giacinto
57cde52b8f
Distributed mode needs an operator to run a NATS cluster. This adds the carrier that replaces its fan-out half, so a deployment eventually needs PostgreSQL and its own HTTP listener and nothing else. pgbus holds one PostgreSQL session per replica, pinned for the life of the process because LISTEN registrations belong to one backend session and a pooled handle would lose them on the next checkout. Publishes go out on the pool with pg_notify. Subjects map onto a channel by their first token, from a closed set of roots. A subject outside the set is refused at publish AND at subscribe rather than mapped to a channel of its own: a channel name is capped at 63 bytes, and one LISTEN per job id would be unbounded. Refused rather than dropped, because a subject that goes nowhere and reports nothing is the class of defect this work exists to remove. PostgreSQL refuses a notify payload of 8000 bytes or more, and several subjects on this bus exceed that in normal operation: a job result carries a whole LLM output, a gallery progress event carries one entry per node. Those are written to a row and the notification carries the id. What is measured against the cap is the ENCODED notification, not the caller's payload, because the subject and the envelope travel too. The filter grammar is not respelled here. Subscribe asks messaging.ValidFilter and delivery asks messaging.SubjectMatches, which makes this the first production caller of a matcher that had only test doubles. New refuses a DSN that names a different database from the pool: that pairing publishes successfully, delivers nothing, on every replica, and reports no error anywhere. Nothing publishes on it and nothing subscribes yet. The construction is wired anyway, because the DSN has exactly one legitimate source and a setting that decides whether any broadcast is delivered should not be invented by whichever call site is migrated first. Delivery is at-most-once, like NATS core. Nothing downstream may read a message it did not receive as evidence about a node: a carrier that cannot deliver is not a worker that is gone. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
81 lines
2.7 KiB
Go
81 lines
2.7 KiB
Go
// SPDX-License-Identifier: MIT
|
|
|
|
package application
|
|
|
|
import (
|
|
"context"
|
|
|
|
. "github.com/onsi/ginkgo/v2"
|
|
. "github.com/onsi/gomega"
|
|
|
|
"github.com/mudler/LocalAI/core/config"
|
|
"github.com/mudler/LocalAI/core/services/pgbus"
|
|
"github.com/mudler/LocalAI/core/services/testutil"
|
|
)
|
|
|
|
// The guard on the one setting that decides whether any broadcast in the
|
|
// deployment is ever delivered.
|
|
//
|
|
// The carrier holds a pinned LISTEN connection opened from a DSN, and publishes
|
|
// travel on a pooled handle opened from another. When those two name different
|
|
// databases every publish succeeds, every subscribe succeeds, and nothing
|
|
// arrives, on every replica, with no error anywhere. There is exactly one
|
|
// legitimate DSN, and these specs are what say so in a way that fails when it
|
|
// stops being true.
|
|
var _ = Describe("opening the deployment's broadcast carrier", func() {
|
|
It("listens on the same database URL the auth pool was built from", func() {
|
|
db, dsn := testutil.SetupTestDBWithDSN()
|
|
cfg := &config.ApplicationConfig{}
|
|
cfg.Auth.DatabaseURL = dsn
|
|
|
|
bus, err := newBroadcastBus(context.Background(), cfg, db)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
DeferCleanup(bus.Close)
|
|
|
|
// Equality with the field, not "is a PostgreSQL URL": the failure being
|
|
// excluded is two databases, and any DSN passes a shape check.
|
|
Expect(bus.DSN()).To(Equal(cfg.Auth.DatabaseURL))
|
|
})
|
|
|
|
It("migrates the spill table, so an oversized broadcast has somewhere to go", func() {
|
|
db, dsn := testutil.SetupTestDBWithDSN()
|
|
cfg := &config.ApplicationConfig{}
|
|
cfg.Auth.DatabaseURL = dsn
|
|
|
|
bus, err := newBroadcastBus(context.Background(), cfg, db)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
DeferCleanup(bus.Close)
|
|
|
|
Expect(db.Migrator().HasTable(&pgbus.BusMessage{})).To(BeTrue())
|
|
})
|
|
|
|
It("refuses to open a carrier whose DSN is not the pool's database", func() {
|
|
db, _ := testutil.SetupTestDBWithDSN()
|
|
_, otherDSN := testutil.SetupTestDBWithDSN()
|
|
cfg := &config.ApplicationConfig{}
|
|
cfg.Auth.DatabaseURL = otherDSN
|
|
|
|
_, err := newBroadcastBus(context.Background(), cfg, db)
|
|
|
|
Expect(err).To(HaveOccurred())
|
|
})
|
|
})
|
|
|
|
var _ = Describe("shutting the distributed services down", func() {
|
|
It("closes the broadcast carrier", func() {
|
|
// A pinned PostgreSQL session and the goroutine parked on it, per
|
|
// replica restart. Nothing else in this process ever closes it, so the
|
|
// line in the shutdown closure is the whole lifecycle.
|
|
db, dsn := testutil.SetupTestDBWithDSN()
|
|
cfg := &config.ApplicationConfig{}
|
|
cfg.Auth.DatabaseURL = dsn
|
|
bus, err := newBroadcastBus(context.Background(), cfg, db)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(bus.IsConnected()).To(BeTrue())
|
|
|
|
(&DistributedServices{Bus: bus}).Shutdown()
|
|
|
|
Expect(bus.IsConnected()).To(BeFalse())
|
|
})
|
|
})
|