fix(11-07): enqueue broadcast jobs before GORM commits a single write

- lighthouse:after_create/update/delete also declare
  Before(gorm:commit_or_rollback_transaction); an After-only anchor put
  them past GORM's own commit, so a plain gdb.Create enqueued its
  broadcast job after the commit on the pool (deferred from 11-05)
- lighthouse gets the testcontainers Postgres harness and TestBroadcastTx
  (commit publishes once, rollback nothing, single-statement write
  enqueues on its own transaction, failed write enqueues nothing)
This commit is contained in:
Jakub Zych
2026-09-30 14:05:17 +02:00
parent c544319cec
commit 6f50b6c940
4 changed files with 521 additions and 5 deletions

View File

@@ -39,7 +39,7 @@ The `centrifugo` sub-package is the Centrifugo driver. It has a hand-rolled `net
- `lighthouse.BroadcastTTLer` or `Binding.TTL` replaces the ttl.
The event name is `{action}.{alias}` lowercased: `lighthouse.ActionCreated`, `lighthouse.ActionUpdated` or `lighthouse.ActionDeleted`, then an alias that defaults to `<plugin>.<model>` (the Go package name, or the parent directory of a `models` package, and the type name). The payload builder receives a `lighthouse.Event` with the action, the `lighthouse.Actor`, the timestamp and the ttl. A soft delete counts as a delete. A delete's channels and payload are computed from a fresh read of the row before it is deleted, so deleting a model that holds only its id still broadcasts. An empty channel list means no broadcast.
- Transactional delivery. GORM callbacks (`lighthouse.CallbackAfterCreate`, `lighthouse.CallbackAfterUpdate`, `lighthouse.CallbackSnapshot` and `lighthouse.CallbackAfterDelete`) are installed through `lagoon.OnDatabase`. They enqueue a `lighthouse.BroadcastArgs` job on the write's `*sql.Tx`, on the `realtime.broadcast_queue` queue with MaxAttempts 1 and the `realtime.broadcast_timeout` timeout. Channel and payload queries and the enqueue run inside a savepoint, so a failure is rolled back to it, logged at Warn with channels and event (never the payload), and the write goes on. A write with a zero primary key, such as `Model(&T{}).Where(…).Updates(…)`, is not broadcast; bulk paths suppress and emit instead. The null driver, or a driver whose `Enabled` reports false (Centrifugo without an API key), gets no jobs.
- Transactional delivery. GORM callbacks (`lighthouse.CallbackAfterCreate`, `lighthouse.CallbackAfterUpdate`, `lighthouse.CallbackSnapshot` and `lighthouse.CallbackAfterDelete`) are installed through `lagoon.OnDatabase`. The after-write callbacks run after the model's own after hook and before GORM commits the transaction it opens for a single-statement write, so they enqueue a `lighthouse.BroadcastArgs` job on the write's `*sql.Tx` in every case (an explicit transaction or a single `Create`, `Save` or `Delete`), on the `realtime.broadcast_queue` queue with MaxAttempts 1 and the `realtime.broadcast_timeout` timeout. Channel and payload queries and the enqueue run inside a savepoint, so a failure is rolled back to it, logged at Warn with channels and event (never the payload), and the write goes on. A write with a zero primary key, such as `Model(&T{}).Where(…).Updates(…)`, is not broadcast; bulk paths suppress and emit instead. The null driver, or a driver whose `Enabled` reports false (Centrifugo without an API key), gets no jobs.
- The broadcast job lowercases the channels and adds the `realtime.broadcast_namespace` prefix unless a channel already has it. It then publishes to one channel or broadcasts to several. A failure is logged as `realtime: broadcast failed` and is not retried. Delivery order across separate jobs is not guaranteed. The payload travels inside the job as a JSON string, so its key order survives Postgres JSONB.
- Suppression: `lighthouse.WithoutBroadcasting` silences one model type for writes made with the context it hands to its function. Other types still broadcast, and a write through an outer context is not suppressed. `lighthouse.Service.Emit` enqueues one `lighthouse.Broadcast` on the caller's transaction and returns its error. Together they turn N row events into one summary event.
- Centrifugo driver (`centrifugo.Driver`, driver name `centrifugo`):

View File

@@ -229,20 +229,26 @@ type defaultPayload struct {
TTL int `json:"ttl"`
}
// commitCallback is GORM's commit of a transaction it opened itself.
const commitCallback = "gorm:commit_or_rollback_transaction"
// installCallbacks registers the broadcast callbacks on gdb, replacing
// earlier ones so a handle shared by several apps broadcasts through the
// most recent service.
// most recent service. The after-write callbacks run after the model's own
// after hook and before GORM commits a single-statement write, so the job
// is enqueued on the write's transaction: GORM appends a callback that
// names only an After anchor to the end of the chain, past the commit.
func (s *Service) installCallbacks(gdb *gorm.DB) error {
cb := gdb.Callback()
if cb.Create().Get(CallbackAfterCreate) == nil {
if err := cb.Create().After("gorm:after_create").Register(CallbackAfterCreate, s.afterCreate); err != nil {
if err := cb.Create().After("gorm:after_create").Before(commitCallback).Register(CallbackAfterCreate, s.afterCreate); err != nil {
return err
}
} else if err := cb.Create().Replace(CallbackAfterCreate, s.afterCreate); err != nil {
return err
}
if cb.Update().Get(CallbackAfterUpdate) == nil {
if err := cb.Update().After("gorm:after_update").Register(CallbackAfterUpdate, s.afterUpdate); err != nil {
if err := cb.Update().After("gorm:after_update").Before(commitCallback).Register(CallbackAfterUpdate, s.afterUpdate); err != nil {
return err
}
} else if err := cb.Update().Replace(CallbackAfterUpdate, s.afterUpdate); err != nil {
@@ -256,7 +262,7 @@ func (s *Service) installCallbacks(gdb *gorm.DB) error {
return err
}
if cb.Delete().Get(CallbackAfterDelete) == nil {
if err := cb.Delete().After("gorm:after_delete").Register(CallbackAfterDelete, s.afterDelete); err != nil {
if err := cb.Delete().After("gorm:after_delete").Before(commitCallback).Register(CallbackAfterDelete, s.afterDelete); err != nil {
return err
}
} else if err := cb.Delete().Replace(CallbackAfterDelete, s.afterDelete); err != nil {

View File

@@ -0,0 +1,362 @@
package lighthouse
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
"sync"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/backpack"
"git.golem15.com/golem15/summercms/modules/compass"
"git.golem15.com/golem15/summercms/modules/conga"
"git.golem15.com/golem15/summercms/modules/lagoon"
"gorm.io/gorm"
)
// Widget broadcasts through its Broadcastable method.
type Widget struct {
ID uint `gorm:"column:id;primaryKey"`
Name string `gorm:"column:name"`
OwnerID uint `gorm:"column:owner_id"`
}
func (Widget) TableName() string { return "acme_widgets" }
// BroadcastChannels publishes to the owner's widget channel; the mixed
// case proves the job lowercases it.
func (w *Widget) BroadcastChannels(context.Context, *gorm.DB) ([]string, error) {
if w.OwnerID == 0 {
return nil, nil
}
return []string{fmt.Sprintf("Widgets:%d", w.OwnerID)}, nil
}
// Gadget is broadcast only through a Binding installed by the test.
type Gadget struct {
ID uint `gorm:"column:id;primaryKey"`
Name string `gorm:"column:name"`
OwnerID uint `gorm:"column:owner_id"`
}
func (Gadget) TableName() string { return "acme_gadgets" }
const broadcastTables = `
CREATE TABLE acme_widgets (id SERIAL PRIMARY KEY, name TEXT NOT NULL, owner_id INT NOT NULL DEFAULT 0);
CREATE TABLE acme_gadgets (id SERIAL PRIMARY KEY, name TEXT NOT NULL UNIQUE, owner_id INT NOT NULL DEFAULT 0);
CREATE TABLE acme_sprockets (id SERIAL PRIMARY KEY, name TEXT NOT NULL, channels TEXT NOT NULL DEFAULT '');
`
// logCapture records log messages with their attributes as strings.
type logCapture struct {
mu sync.Mutex
records []capturedLog
}
type capturedLog struct {
level slog.Level
msg string
attrs map[string]string
}
func (h *logCapture) Enabled(context.Context, slog.Level) bool { return true }
func (h *logCapture) WithAttrs([]slog.Attr) slog.Handler { return h }
func (h *logCapture) WithGroup(string) slog.Handler { return h }
func (h *logCapture) Handle(_ context.Context, r slog.Record) error {
rec := capturedLog{level: r.Level, msg: r.Message, attrs: map[string]string{}}
r.Attrs(func(a slog.Attr) bool {
rec.attrs[a.Key] = a.Value.String()
return true
})
h.mu.Lock()
h.records = append(h.records, rec)
h.mu.Unlock()
return nil
}
func (h *logCapture) count(msg string) int {
h.mu.Lock()
defer h.mu.Unlock()
n := 0
for _, r := range h.records {
if r.msg == msg {
n++
}
}
return n
}
func (h *logCapture) find(msg string) (capturedLog, bool) {
h.mu.Lock()
defer h.mu.Unlock()
for _, r := range h.records {
if r.msg == msg {
return r, true
}
}
return capturedLog{}, false
}
func (h *logCapture) all() string {
h.mu.Lock()
defer h.mu.Unlock()
var b strings.Builder
for _, r := range h.records {
fmt.Fprintf(&b, "%s %s %v\n", r.level, r.msg, r.attrs)
}
return b.String()
}
// lhEnv is an app with the memory driver (unless kv names another), the
// broadcast callbacks installed through the production boot order
// (From before the database is published) and a running worker.
type lhEnv struct {
app *backpack.App
svc *Service
db *sql.DB
gdb *gorm.DB
logs *logCapture
}
func newLHEnv(t *testing.T, kv map[string]any, setup func(t *testing.T, svc *Service)) lhEnv {
t.Helper()
db, dsn := migratedDB(t)
ctx := t.Context()
cfg, err := compass.Open(compass.Options{Dir: t.TempDir(), Env: "testing", Environ: []string{}})
if err != nil {
t.Fatal(err)
}
settings := map[string]any{"database.dsn": dsn, "realtime.driver": "memory"}
for k, v := range kv {
settings[k] = v
}
for k, v := range settings {
if err := cfg.Set(k, v); err != nil {
t.Fatal(err)
}
}
app := backpack.New(cfg)
logs := &logCapture{}
if err := app.Publish(slog.New(logs)); err != nil {
t.Fatal(err)
}
svc, err := From(app)
if err != nil {
t.Fatal(err)
}
if setup != nil {
setup(t, svc)
}
gdb, err := lagoon.Use(ctx, db)
if err != nil {
t.Fatal(err)
}
if err := gdb.Exec(broadcastTables).Error; err != nil {
t.Fatal(err)
}
if err := lagoon.Publish(app, db, gdb); err != nil {
t.Fatal(err)
}
w, err := conga.StartWorker(ctx, app, nil, conga.WorkerOptions{})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := w.Stop(stopCtx); err != nil {
t.Errorf("stop worker: %v", err)
}
})
return lhEnv{app: app, svc: svc, db: db, gdb: gdb, logs: logs}
}
func (e lhEnv) memory(t *testing.T) *MemoryDriver {
t.Helper()
m, ok := e.svc.Driver().(*MemoryDriver)
if !ok {
t.Fatalf("driver is %T, want the memory driver", e.svc.Driver())
}
return m
}
// broadcastJobs counts summer.broadcast jobs through a separate connection.
func (e lhEnv) broadcastJobs(t *testing.T) int {
t.Helper()
var n int
if err := e.db.QueryRowContext(t.Context(), `SELECT count(*) FROM river_job WHERE kind = 'summer.broadcast'`).Scan(&n); err != nil {
t.Fatal(err)
}
return n
}
// waitPublications waits for exactly n publications after the first from,
// then waits a little longer to catch extra ones.
func waitPublications(t *testing.T, m *MemoryDriver, from, n int) []Publication {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for len(m.Publications()) < from+n {
if time.Now().After(deadline) {
t.Fatalf("got %d publications, want %d", len(m.Publications())-from, n)
}
time.Sleep(20 * time.Millisecond)
}
time.Sleep(300 * time.Millisecond)
pubs := m.Publications()[from:]
if len(pubs) != n {
t.Fatalf("got %d publications, want exactly %d: %+v", len(pubs), n, pubs)
}
return pubs
}
// payloadKeys returns the top-level keys of a JSON object in order.
func payloadKeys(t *testing.T, raw json.RawMessage) []string {
t.Helper()
dec := json.NewDecoder(strings.NewReader(string(raw)))
if tok, err := dec.Token(); err != nil || tok != json.Delim('{') {
t.Fatalf("payload %s is not an object", raw)
}
var keys []string
for dec.More() {
tok, err := dec.Token()
if err != nil {
t.Fatal(err)
}
keys = append(keys, tok.(string))
var skip json.RawMessage
if err := dec.Decode(&skip); err != nil {
t.Fatal(err)
}
}
return keys
}
// TestBroadcastTx covers RT-03 and T-11-05: model broadcasts are River jobs
// enqueued in the write's transaction, published once after the commit and
// never after a rollback, including single-statement writes for which GORM
// opens its own transaction.
func TestBroadcastTx(t *testing.T) {
var (
mu sync.Mutex
inTxLog []bool
)
env := newLHEnv(t, nil, func(t *testing.T, svc *Service) {
err := Bind[Gadget](svc, Binding[Gadget]{
Alias: "acme.gadget",
Channels: func(_ context.Context, tx *gorm.DB, m *Gadget) ([]string, error) {
_, inTx := tx.Statement.ConnPool.(*sql.Tx)
mu.Lock()
inTxLog = append(inTxLog, inTx)
mu.Unlock()
return []string{fmt.Sprintf("gadgets:%d", m.OwnerID)}, nil
},
})
if err != nil {
t.Fatal(err)
}
})
mem := env.memory(t)
ctx := t.Context()
t.Run("commit_publishes_once_after_commit", func(t *testing.T) {
from := len(mem.Publications())
before := env.broadcastJobs(t)
err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
if err := tx.WithContext(ctx).Create(&Widget{Name: "w1", OwnerID: 5}).Error; err != nil {
return err
}
var inside int
if err := tx.Raw(`SELECT count(*) FROM river_job WHERE kind = 'summer.broadcast'`).Scan(&inside).Error; err != nil {
return err
}
if inside != before+1 {
t.Errorf("jobs inside the write transaction = %d, want %d", inside, before+1)
}
if outside := env.broadcastJobs(t); outside != before {
t.Errorf("job visible outside the transaction before commit: %d, want %d", outside, before)
}
return nil
})
if err != nil {
t.Fatal(err)
}
p := waitPublications(t, mem, from, 1)[0]
if p.Method != "publish" || len(p.Channels) != 1 || p.Channels[0] != "widgets:5" {
t.Fatalf("publication = %s %v, want publish [widgets:5]", p.Method, p.Channels)
}
if p.Event != "created.lighthouse.widget" {
t.Fatalf("event = %q", p.Event)
}
if got := strings.Join(payloadKeys(t, p.Payload), ","); got != "model,actor,timestamp,ttl" {
t.Fatalf("payload keys = %s", got)
}
var body struct {
Model Widget `json:"model"`
Actor Actor `json:"actor"`
TTL int `json:"ttl"`
}
if err := json.Unmarshal(p.Payload, &body); err != nil {
t.Fatal(err)
}
if body.Model.Name != "w1" || body.TTL != DefaultTTL || body.Actor.UserID != nil || body.Actor.Name == nil || *body.Actor.Name != "System" {
t.Fatalf("payload = %s", p.Payload)
}
})
t.Run("rollback_publishes_nothing", func(t *testing.T) {
from := len(mem.Publications())
before := env.broadcastJobs(t)
rollback := errors.New("rollback")
err := lagoon.Transaction(ctx, env.gdb, func(ctx context.Context, tx *gorm.DB) error {
if err := tx.WithContext(ctx).Create(&Widget{Name: "w-rolled-back", OwnerID: 5}).Error; err != nil {
return err
}
return rollback
})
if !errors.Is(err, rollback) {
t.Fatalf("err = %v", err)
}
if after := env.broadcastJobs(t); after != before {
t.Fatalf("broadcast jobs %d -> %d after a rollback", before, after)
}
time.Sleep(300 * time.Millisecond)
if n := len(mem.Publications()) - from; n != 0 {
t.Fatalf("%d publications after a rollback", n)
}
})
t.Run("single_statement_write_enqueues_in_its_own_transaction", func(t *testing.T) {
from := len(mem.Publications())
mu.Lock()
inTxLog = nil
mu.Unlock()
if err := env.gdb.WithContext(ctx).Create(&Gadget{Name: "g1", OwnerID: 3}).Error; err != nil {
t.Fatal(err)
}
mu.Lock()
got := append([]bool(nil), inTxLog...)
mu.Unlock()
if len(got) != 1 || !got[0] {
t.Fatalf("channels ran on the write's transaction: %v, want [true] (the job must be enqueued before GORM commits)", got)
}
p := waitPublications(t, mem, from, 1)[0]
if p.Event != "created.acme.gadget" || p.Channels[0] != "gadgets:3" {
t.Fatalf("publication = %+v", p)
}
})
t.Run("failed_single_statement_write_enqueues_nothing", func(t *testing.T) {
before := env.broadcastJobs(t)
if err := env.gdb.WithContext(ctx).Create(&Gadget{Name: "g1", OwnerID: 3}).Error; err == nil {
t.Fatal("duplicate gadget was created")
}
if after := env.broadcastJobs(t); after != before {
t.Fatalf("broadcast jobs %d -> %d after a failed write", before, after)
}
})
}

View File

@@ -0,0 +1,148 @@
package lighthouse
import (
"context"
"database/sql"
"fmt"
"net/url"
"os"
"strings"
"sync/atomic"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/lagoon"
_ "github.com/jackc/pgx/v5/stdlib"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/postgres"
)
var (
lhPG *postgres.PostgresContainer
lhSQL *sql.DB
lhDSN string
lhPGErr error
dbSeq atomic.Int64
)
func TestMain(m *testing.M) {
code := 1
if !testShort() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
lhPGErr = startLighthousePostgres(ctx)
cancel()
if lhPGErr != nil {
fmt.Fprintf(os.Stderr, "lighthouse: testcontainers postgres: %v\n", lhPGErr)
stopLighthousePostgres()
os.Exit(1)
}
}
code = m.Run()
stopLighthousePostgres()
os.Exit(code)
}
func testShort() bool {
for _, a := range os.Args {
if a == "-test.short" {
return true
}
}
return false
}
func startLighthousePostgres(ctx context.Context) error {
ctr, err := postgres.Run(ctx,
"postgres:16-alpine",
postgres.WithDatabase("lighthouse"),
postgres.WithUsername("lighthouse"),
postgres.WithPassword("lighthouse"),
postgres.BasicWaitStrategies(),
testcontainers.WithEnv(map[string]string{
"POSTGRES_INITDB_ARGS": "--locale-provider=icu --icu-locale=pl-PL --encoding=UTF8",
}),
)
if err != nil {
return err
}
lhPG = ctr
dsn, err := ctr.ConnectionString(ctx, "sslmode=disable")
if err != nil {
return err
}
db, err := sql.Open("pgx", dsn)
if err != nil {
return err
}
if err := db.PingContext(ctx); err != nil {
_ = db.Close()
return err
}
lhSQL = db
lhDSN = dsn
return nil
}
func stopLighthousePostgres() {
if lhSQL != nil {
_ = lhSQL.Close()
}
if lhPG != nil {
_ = testcontainers.TerminateContainer(lhPG)
}
}
func adminDB(t *testing.T) *sql.DB {
t.Helper()
if testing.Short() {
t.Skip("requires testcontainers postgres")
}
if lhPGErr != nil {
t.Fatalf("postgres unavailable: %v", lhPGErr)
}
if lhSQL == nil {
t.Fatal("postgres unavailable: container was not started")
}
return lhSQL
}
// migratedDB returns a dedicated ICU pl-PL database migrated with
// lagoon.Migrate (River v7 and summer_jobs) plus its DSN.
func migratedDB(t *testing.T) (*sql.DB, string) {
t.Helper()
admin := adminDB(t)
ctx := t.Context()
name := fmt.Sprintf("lighthouse_%d", dbSeq.Add(1))
if _, err := admin.ExecContext(ctx, `CREATE DATABASE `+name+` TEMPLATE template0 ENCODING 'UTF8' LOCALE_PROVIDER icu ICU_LOCALE 'pl-PL'`); err != nil && !strings.Contains(err.Error(), "already exists") {
t.Fatalf("create %s: %v", name, err)
}
dsn, err := dsnWithDB(lhDSN, name)
if err != nil {
t.Fatal(err)
}
db, err := sql.Open("pgx", dsn)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_ = db.Close()
_, _ = admin.ExecContext(context.Background(), `DROP DATABASE IF EXISTS `+name+` WITH (FORCE)`)
})
gdb, err := lagoon.Use(ctx, db)
if err != nil {
t.Fatal(err)
}
if err := lagoon.Migrate(gdb, nil); err != nil {
t.Fatal(err)
}
return db, dsn
}
func dsnWithDB(dsn, name string) (string, error) {
u, err := url.Parse(dsn)
if err != nil {
return "", err
}
u.Path = "/" + name
return u.String(), nil
}