mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-22 14:14:54 -04:00
Ettore Di Giacinto
5562cd7ad3
PostgreSQL holds undelivered notifications in a queue it shares with every session on the server, and it kills a listener that stops draining. Two failure modes follow, and both are silent: a carrier that blocked on a slow resolver would lose its connection and with it every later broadcast, and a carrier that reconnected without re-registering would be connected and deaf. The receive and dispatch halves were already separate. What was missing is everything around them. The listener path moves into listener.go and gains a carrier-level Dropped() so a replica that is behind can be seen; the queue depth and the spill retention become Config fields with exported defaults; the LISTEN session gets an application_name so an operator can count listeners in pg_stat_activity and a spec can drop exactly one of them; and OnReconnect fires after the re-LISTEN, on a goroutine of its own, because a callback re-hydrates from a database and must never run on the path whose only job is to drain. That callback is reached through an optional interface assertion, so deleting its invocation compiles and every adopter silently stops converging. The spec is the only guard, and it is named in a comment at the site. The slow consumer is proved through the transport rather than a seam: an ACCESS EXCLUSIVE lock on bus_messages stalls the resolver's spill SELECT for exactly as long as the spec holds it, and the listener is shown still draining and dropping while it does. The dropped connection is a pg_terminate_backend matched on the carrier's own application name. Neither Dropped nor IsConnected is on messaging.Broadcaster, and a spec asserts that over the interface type. Both are facts about a frontend; the conditions a scheduler acts on are facts about a worker, and no consumer holding the interface can read one as the other. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
64 lines
2.6 KiB
Go
64 lines
2.6 KiB
Go
// SPDX-License-Identifier: MIT
|
|
|
|
package pgbus
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// SpillSweepSQL retires spilled broadcasts older than the retention it is given.
|
|
//
|
|
// The cutoff is computed by the DATABASE and never in Go, and that is the whole
|
|
// point of holding the statement here where a spec can read it. Replicas do not
|
|
// share a clock: a replica running minutes ahead would compute a cutoff in the
|
|
// future and delete rows its peers have not read yet, and a replica running
|
|
// behind would never delete anything. now() is the one clock every replica
|
|
// agrees on, because it is the clock the rows were written by.
|
|
//
|
|
// PostgreSQL-only, like everything else in this package. New refuses a handle on
|
|
// any other dialect, which is where that is guarded.
|
|
const SpillSweepSQL = "DELETE FROM bus_messages WHERE created_at < now() - make_interval(secs => ?)"
|
|
|
|
// BusMessage is one broadcast too large to travel in a notification.
|
|
type BusMessage struct {
|
|
ID string `gorm:"primaryKey;size:36"`
|
|
Subject string `gorm:"size:255;index"`
|
|
Payload []byte
|
|
CreatedAt time.Time `gorm:"index"`
|
|
}
|
|
|
|
// Migrate creates the spill table. It is separate from New because a deployment
|
|
// migrates once at boot while every replica opens a carrier.
|
|
func Migrate(ctx context.Context, db *gorm.DB) error {
|
|
return db.WithContext(ctx).AutoMigrate(&BusMessage{})
|
|
}
|
|
|
|
// SweepSpill deletes spilled broadcasts older than retention.
|
|
//
|
|
// For a caller that has a handle but no carrier, which is why it takes a
|
|
// *gorm.DB. The carrier's own purge loop calls PurgeBefore instead, because it
|
|
// needs the count; both run the one statement in SpillSweepSQL.
|
|
func SweepSpill(ctx context.Context, db *gorm.DB, retention time.Duration) error {
|
|
return db.WithContext(ctx).Exec(SpillSweepSQL, retention.Seconds()).Error
|
|
}
|
|
|
|
// PurgeBefore deletes this carrier's spilled rows older than olderThan and
|
|
// returns how many it deleted.
|
|
//
|
|
// The count is the point. SweepSpill answers "did the statement run", which a
|
|
// purge loop that is retiring nothing answers just as happily, and a spill table
|
|
// that grows without bound is a disk that fills long after the change that
|
|
// caused it. This is also what the purge loop calls, so the count an operator
|
|
// can ask for and the deletion the carrier performs are the same statement.
|
|
func (b *Bus) PurgeBefore(ctx context.Context, olderThan time.Duration) (int64, error) {
|
|
res := b.cfg.DB.WithContext(ctx).Exec(SpillSweepSQL, olderThan.Seconds())
|
|
if res.Error != nil {
|
|
return 0, fmt.Errorf("pgbus: retiring spilled broadcasts: %w", res.Error)
|
|
}
|
|
return res.RowsAffected, nil
|
|
}
|