From 112527db833cfe0e9d45c59d7960d3ec42f702cc Mon Sep 17 00:00:00 2001 From: DiheChen Date: Wed, 11 Nov 2026 00:05:35 +0800 Subject: [PATCH] feat: write listening history as songs start and members join --- internal/adapter/in/api/api_test.go | 2 +- internal/adapter/in/api/room_test.go | 2 +- internal/adapter/in/api/ws_test.go | 2 +- internal/app/admin/admin_test.go | 2 +- internal/app/identity/identity_test.go | 4 +- internal/app/lobby/lobby_test.go | 2 +- internal/app/realtime/cluster_test.go | 8 +- internal/app/realtime/context_test.go | 2 +- internal/app/realtime/gateway.go | 4 +- internal/app/realtime/gateway_test.go | 4 +- internal/app/realtime/host.go | 16 ++- internal/app/realtime/host_test.go | 136 +++++++++++++++++++++++-- internal/app/realtime/observer_test.go | 4 +- internal/bootstrap/public.go | 2 +- 14 files changed, 163 insertions(+), 27 deletions(-) diff --git a/internal/adapter/in/api/api_test.go b/internal/adapter/in/api/api_test.go index 2fef4f0..d4f5aa0 100644 --- a/internal/adapter/in/api/api_test.go +++ b/internal/adapter/in/api/api_test.go @@ -39,7 +39,7 @@ const allowedOrigin = "https://allowed.example" func services(store *porttest.Store, rooms port.RoomDirectory, adminToken string) Services { music := newCatalog(rooms, 10) node := porttest.NewRedis().Node("a") - _, gateway := realtime.New(store, rooms, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + _, gateway := realtime.New(store, rooms, store, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) l := lobby.New(rooms, node) return Services{ Identity: identity.New(store, rooms, store, gateway), diff --git a/internal/adapter/in/api/room_test.go b/internal/adapter/in/api/room_test.go index 5448c89..5f1a731 100644 --- a/internal/adapter/in/api/room_test.go +++ b/internal/adapter/in/api/room_test.go @@ -43,7 +43,7 @@ func newRoomFixture(t *testing.T) roomFixture { store := porttest.New() music := newCatalog(store, 10) node := porttest.NewRedis().Node("a") - host, gateway := realtime.New(store, store, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + host, gateway := realtime.New(store, store, store, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := realtime.Start(ctx, host, gateway) if err != nil { diff --git a/internal/adapter/in/api/ws_test.go b/internal/adapter/in/api/ws_test.go index 4d736ed..702ad40 100644 --- a/internal/adapter/in/api/ws_test.go +++ b/internal/adapter/in/api/ws_test.go @@ -134,7 +134,7 @@ func newWSFixtureWithTickets(t *testing.T, store *porttest.Store, tickets port.T music := newCatalog(directory, 10) node := porttest.NewRedis().Node("a") preparer := porttest.NewPreparer() - host, gateway := realtime.New(snapshots, store, store, music, preparer, realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + host, gateway := realtime.New(snapshots, store, store, store, music, preparer, realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := realtime.Start(ctx, host, gateway) if err != nil { diff --git a/internal/app/admin/admin_test.go b/internal/app/admin/admin_test.go index b22a18f..ff90c30 100644 --- a/internal/app/admin/admin_test.go +++ b/internal/app/admin/admin_test.go @@ -34,7 +34,7 @@ func newService(t *testing.T, token string) fixture { t.Helper() store := porttest.New() node := porttest.NewRedis().Node("a") - host, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + host, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := realtime.Start(ctx, host, gateway) if err != nil { diff --git a/internal/app/identity/identity_test.go b/internal/app/identity/identity_test.go index 35387c1..914dd84 100644 --- a/internal/app/identity/identity_test.go +++ b/internal/app/identity/identity_test.go @@ -28,7 +28,7 @@ func newService(t *testing.T) (*Service, *porttest.Store) { // 没有调用 realtime.Start：这些测试不运行房间。 func newGateway(store *porttest.Store, node *porttest.Node, bus port.Bus) *realtime.Gateway { - _, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler)) + _, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler)) return gateway } @@ -206,7 +206,7 @@ func TestEnterRoomLeavesWhenTicketIsRevokedWhileJoining(t *testing.T) { synctest.Test(t, func(t *testing.T) { store := porttest.New() node := porttest.NewRedis().Node("a") - host, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + host, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(t.Context()) wait, err := realtime.Start(ctx, host, gateway) if err != nil { diff --git a/internal/app/lobby/lobby_test.go b/internal/app/lobby/lobby_test.go index 1fa7bb8..3d7caad 100644 --- a/internal/app/lobby/lobby_test.go +++ b/internal/app/lobby/lobby_test.go @@ -14,7 +14,7 @@ import ( func newService(t *testing.T, store *porttest.Store) (*Service, *realtime.Gateway) { t.Helper() node := porttest.NewRedis().Node("a") - host, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + host, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := realtime.Start(ctx, host, gateway) if err != nil { diff --git a/internal/app/realtime/cluster_test.go b/internal/app/realtime/cluster_test.go index 5e62771..b0a1a92 100644 --- a/internal/app/realtime/cluster_test.go +++ b/internal/app/realtime/cluster_test.go @@ -169,7 +169,7 @@ func newFlakyBusInstance(t *testing.T, store *porttest.Store, logger *slog.Logge songs := newSongs() node := porttest.NewRedis().Node("a") bus := &flakyBus{Bus: node} - h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, logger) + h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, logger) return h, g, node, bus } @@ -381,7 +381,7 @@ func TestGatewayRejoinsAfterItsAliveKeyExpired(t *testing.T) { a := startInstance(t, redis, "a", store, store, songs) node := redis.Node("b") liveness := &flakyLiveness{Liveness: node} - h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "b", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler)) + h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "b", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := Start(ctx, h, g) if err != nil { @@ -457,7 +457,7 @@ func startFlakyInstance(t *testing.T, redis *porttest.Redis, id string, snapshot t.Helper() node := redis.Node(id) leases, bus := &flakyOwners{Leases: node}, &flakyForwards{Bus: node} - h, g := New(snapshots, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: id, Leases: leases, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler)) + h, g := New(snapshots, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: id, Leases: leases, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := Start(ctx, h, g) if err != nil { @@ -722,7 +722,7 @@ func TestCommandFollowsOwnerAfterLosingTakeover(t *testing.T) { node := f.redis.Node("c") leases := &staleLeases{Leases: node} songs := newSongs() - h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "c", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "c", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := Start(ctx, h, g) if err != nil { diff --git a/internal/app/realtime/context_test.go b/internal/app/realtime/context_test.go index 1a5bbba..4c45325 100644 --- a/internal/app/realtime/context_test.go +++ b/internal/app/realtime/context_test.go @@ -76,7 +76,7 @@ func TestPreparationKeepsRequestValues(t *testing.T) { node := redis.Node("a") preparer := contextPreparer{started: make(chan context.Context, 1)} f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}}, - newHost(store, store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler))) + newHost(store, store, store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler))) id := f.createRoom(t, room.Persistent) ctx, cancel := context.WithCancel(context.WithValue(context.Background(), requestValue{}, "点歌")) diff --git a/internal/app/realtime/gateway.go b/internal/app/realtime/gateway.go index fa46f7f..86fdc23 100644 --- a/internal/app/realtime/gateway.go +++ b/internal/app/realtime/gateway.go @@ -87,8 +87,8 @@ type roomLink struct { confirmed uint64 } -func New(snapshots port.RoomSnapshots, directory port.RoomDirectory, tickets port.ExistingTickets, music *catalog.Service, preparer port.Preparer, cluster Cluster, logger *slog.Logger) (*Host, *Gateway) { - h := newHost(snapshots, directory, preparer, music, cluster, logger) +func New(snapshots port.RoomSnapshots, directory port.RoomDirectory, listens port.ListenRecorder, tickets port.ExistingTickets, music *catalog.Service, preparer port.Preparer, cluster Cluster, logger *slog.Logger) (*Host, *Gateway) { + h := newHost(snapshots, directory, listens, preparer, music, cluster, logger) g := &Gateway{ self: cluster.Instance, host: h, diff --git a/internal/app/realtime/gateway_test.go b/internal/app/realtime/gateway_test.go index 97f72ed..9e6a2f9 100644 --- a/internal/app/realtime/gateway_test.go +++ b/internal/app/realtime/gateway_test.go @@ -43,7 +43,7 @@ func startInstance(t *testing.T, redis *porttest.Redis, id string, snapshots por func startLoggingInstance(t *testing.T, redis *porttest.Redis, id string, snapshots port.RoomSnapshots, store *porttest.Store, tickets port.ExistingTickets, songs *songs, logger *slog.Logger) *instance { t.Helper() node := redis.Node(id) - h, g := New(snapshots, store, tickets, songs.music, songs.preparer, cluster(node, id), logger) + h, g := New(snapshots, store, store, tickets, songs.music, songs.preparer, cluster(node, id), logger) ctx, stop := context.WithCancel(context.Background()) wait, err := Start(ctx, h, g) if err != nil { @@ -603,7 +603,7 @@ func startCountingGateway(t *testing.T, store *porttest.Store) (*Gateway, *portt songs := newSongs() c := cluster(node, "a") c.Bus = bus - h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, c, slog.New(slog.DiscardHandler)) + h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, c, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := Start(ctx, h, g) if err != nil { diff --git a/internal/app/realtime/host.go b/internal/app/realtime/host.go index 05f9b17..4a83409 100644 --- a/internal/app/realtime/host.go +++ b/internal/app/realtime/host.go @@ -49,6 +49,7 @@ type Host struct { self string snapshots port.RoomSnapshots directory port.RoomDirectory + listens port.ListenRecorder preparer port.Preparer catalog *catalog.Service leases port.Leases @@ -120,11 +121,12 @@ type hostedRoom struct { timers map[room.Token]*time.Timer } -func newHost(snapshots port.RoomSnapshots, directory port.RoomDirectory, preparer port.Preparer, music *catalog.Service, cluster Cluster, logger *slog.Logger) *Host { +func newHost(snapshots port.RoomSnapshots, directory port.RoomDirectory, listens port.ListenRecorder, preparer port.Preparer, music *catalog.Service, cluster Cluster, logger *slog.Logger) *Host { return &Host{ self: cluster.Instance, snapshots: snapshots, directory: directory, + listens: listens, preparer: preparer, catalog: music, leases: cluster.Leases, @@ -592,6 +594,8 @@ func (h *Host) execute(ctx context.Context, r *hostedRoom, effects []room.Effect r.info = e.Snapshot.Info case room.RecordPlay: // 已经和 Persist 在同一个事务中写入。 + case room.RecordListen: + h.recordListen(ctx, e) case room.Emit: if m, ok := e.Event.(room.MembersChanged); ok && !h.setOnline(ctx, r, len(m.Sessions)) { return false @@ -679,6 +683,16 @@ func (h *Host) persist(ctx context.Context, r *hostedRoom, s room.Snapshot, play return h.snapshots.SaveRoom(ctx, r.term, s, plays) } +// 播放历史不是房间的状态：写入失败不放弃房间。与落盘一样不随 ctx 取消。 +func (h *Host) recordListen(ctx context.Context, l room.RecordListen) { + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), storeTimeout) + defer cancel() + listen := port.Listen{Track: l.Entry.Track, Meta: l.Entry.Meta, HeardAt: l.StartedAt} + if err := h.listens.RecordListen(writeCtx, l.Members, listen); err != nil { + h.logger.WarnContext(ctx, "写入播放历史失败", "error", err) + } +} + func (h *Host) remove(ctx context.Context, r *hostedRoom) error { ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), storeTimeout) defer cancel() diff --git a/internal/app/realtime/host_test.go b/internal/app/realtime/host_test.go index 8af45e0..519187c 100644 --- a/internal/app/realtime/host_test.go +++ b/internal/app/realtime/host_test.go @@ -161,7 +161,7 @@ func newHostFixture(t *testing.T, snapshots port.RoomSnapshots, store *porttest. redis := porttest.NewRedis() node := redis.Node("a") return startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}}, - newHost(snapshots, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.DiscardHandler))) + newHost(snapshots, store, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.DiscardHandler))) } func startHost(t *testing.T, f *hostFixture, h *Host) *hostFixture { @@ -289,6 +289,128 @@ func TestPersistFailureAbandonsRoom(t *testing.T) { }) } +// 写入时把此前发布的房间消息和这次写入按发生的先后记进同一个日志。 +type orderedListens struct { + *porttest.Store + spy spy + fail bool + + mu sync.Mutex + log []string +} + +func (l *orderedListens) RecordListen(ctx context.Context, memberIDs []string, listen port.Listen) error { + l.mu.Lock() + defer l.mu.Unlock() + for _, m := range l.spy.take() { + l.log = append(l.log, describe(m)) + } + l.log = append(l.log, fmt.Sprintf("listen %d %s", len(memberIDs), listen.Track.ID)) + if l.fail { + return errors.New("连接中断") + } + return l.Store.RecordListen(ctx, memberIDs, listen) +} + +func (l *orderedListens) messages() []string { + l.mu.Lock() + defer l.mu.Unlock() + for _, m := range l.spy.take() { + l.log = append(l.log, describe(m)) + } + log := l.log + l.log = nil + return log +} + +func newListenFixture(t *testing.T, fail bool, logger *slog.Logger) (*hostFixture, *orderedListens) { + t.Helper() + store := porttest.New() + songs := newSongs() + songs.add(sunny, sunnyM, sunnyP) + redis := porttest.NewRedis() + node := redis.Node("a") + listens := &orderedListens{Store: store, spy: spy{redis.Node("spy")}, fail: fail} + f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: listens.spy}, + newHost(store, store, listens, songs.preparer, songs.music, cluster(node, "a"), logger)) + return f, listens +} + +func createMembers(t *testing.T, store *porttest.Store, names ...string) []member.Member { + t.Helper() + members := make([]member.Member, len(names)) + for i, name := range names { + m, err := store.CreateMember(t.Context(), []byte(name), name) + if err != nil { + t.Fatal(err) + } + members[i] = m + } + return members +} + +func checkLog(t *testing.T, got []string, want ...string) { + t.Helper() + if !slices.Equal(got, want) { + t.Errorf("房间消息和写入的先后：\n%s\n期望：\n%s", strings.Join(got, "\n"), strings.Join(want, "\n")) + } +} + +func TestListensAreWrittenBeforeTheBroadcast(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + f, listens := newListenFixture(t, false, slog.New(slog.DiscardHandler)) + members := createMembers(t, f.store, "alice", "bob") + a, b := members[0], members[1] + id := f.createRoom(t, room.Persistent) + start := time.Now() + + f.send(t, id, room.Join{Session: "s1", Member: a}) + f.send(t, id, command("r1", a, room.Enqueue{Track: sunny, Meta: sunnyM})) + checkLog(t, listens.messages(), + "started 1", + "1/1 room.MembersChanged", + "1/2 room.Activity", + "1/3 room.QueueChanged", + "reply r1 room.EnqueueResult", + "listen 1 186016", + "1/4 room.QueueChanged", + "1/5 room.PlaybackChanged", + ) + f.send(t, id, room.Join{Session: "s2", Member: b}) + checkLog(t, listens.messages(), "listen 1 186016", "1/6 room.MembersChanged") + + startedAt := start.Add(200 * time.Millisecond) + for _, m := range members { + got, err := f.store.MemberListens(t.Context(), m.ID) + if err != nil || len(got) != 1 || got[0].Track != sunny || got[0].Meta.Title != sunnyM.Title || !got[0].HeardAt.Equal(startedAt) { + t.Errorf("%s 的历史 = %+v, %v，期望《晴天》，时间是开播的 %v", m.Name, got, err, startedAt) + } + } + }) +} + +func TestListenFailureKeepsRoomRunning(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var logs logBuffer + f, listens := newListenFixture(t, true, slog.New(slog.NewJSONHandler(&logs, nil))) + a := createMembers(t, f.store, "alice")[0] + id := f.createRoom(t, room.Persistent) + + f.send(t, id, room.Join{Session: "s1", Member: a}) + f.send(t, id, command("r1", a, room.Enqueue{Track: sunny, Meta: sunnyM})) + if got := listens.messages(); !slices.Contains(got, "1/5 room.PlaybackChanged") { + t.Errorf("写入失败之后的房间消息 = %v，期望照常广播开播", got) + } + if owner := f.owner(t, id); owner != "a" { + t.Errorf("写入失败之后房间的 owner = %q，期望不放弃房间", owner) + } + entries := logs.entries(t, "写入播放历史失败") + if len(entries) != 1 || entries[0]["level"] != "WARN" { + t.Errorf("写入失败的日志 = %v，期望一条 WARN", entries) + } + }) +} + // 每次 Restore 的 token 都从 1 开始：放弃房间时如果不停下计时器，它会提前关闭重新接管后的房间。 func TestTimersOfAbandonedRoomAreDiscarded(t *testing.T) { synctest.Test(t, func(t *testing.T) { @@ -341,7 +463,7 @@ func TestPrepareOfAbandonedRunIsDiscarded(t *testing.T) { redis := porttest.NewRedis() node := redis.Node("a") f := startHost(t, &hostFixture{store: store.Store, redis: redis, node: node, spy: spy{redis.Node("spy")}}, - newHost(store, store.Store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler))) + newHost(store, store.Store, store.Store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler))) id := f.createRoom(t, room.Persistent) f.send(t, id, command("r1", alice, room.Enqueue{Track: sunny, Meta: sunnyM})) f.spy.take() @@ -461,7 +583,7 @@ func TestHostAcceptsRoomsOnlyBetweenStartAndShutdown(t *testing.T) { songs := newSongs() redis := porttest.NewRedis() watcher := spy{redis.Node("spy")} - h := newHost(store, store, songs.preparer, songs.music, cluster(redis.Node("a"), "a"), slog.New(slog.DiscardHandler)) + h := newHost(store, store, store, songs.preparer, songs.music, cluster(redis.Node("a"), "a"), slog.New(slog.DiscardHandler)) id, err := store.CreateRoom(t.Context(), port.NewRoom{Kind: room.Persistent, Name: "房间"}) if err != nil { t.Fatal(err) @@ -673,7 +795,7 @@ func newFlakyFixture(t *testing.T) (*hostFixture, *flakyLeases) { node := redis.Node("a") leases := &flakyLeases{Leases: node, resume: make(chan struct{})} f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}}, - newHost(store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))) + newHost(store, store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))) return f, leases } @@ -876,7 +998,7 @@ func TestFailedReleaseNoticeIsLogged(t *testing.T) { node := redis.Node("a") var logs logBuffer f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}}, - newHost(store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: noReleaseNotice{node}}, slog.New(logctx.NewHandler(slog.NewJSONHandler(&logs, nil))))) + newHost(store, store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: noReleaseNotice{node}}, slog.New(logctx.NewHandler(slog.NewJSONHandler(&logs, nil))))) id := f.createRoom(t, room.Persistent) f.send(t, id, room.Join{Session: "s1", Member: alice}) if err := f.host.Shutdown(t.Context()); err != nil { @@ -1000,7 +1122,7 @@ func TestTakeoverFailuresAreLoggedWhenTheyStart(t *testing.T) { node := redis.Node("a") var logs logBuffer f := startHost(t, &hostFixture{store: store.Store, redis: redis, node: node, spy: spy{redis.Node("spy")}}, - newHost(store, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.NewJSONHandler(&logs, nil)))) + newHost(store, store, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.NewJSONHandler(&logs, nil)))) for range 3 { f.createRoom(t, room.Persistent) } @@ -1087,7 +1209,7 @@ func TestRecoveredGatewayKeepsItsSessions(t *testing.T) { gateway, unreachable := redis.Node("b"), redis.Node("c") liveness := &recoveringLiveness{Liveness: node, gateway: gateway, unreachable: "c"} f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}}, - newHost(store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler))) + newHost(store, store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler))) id := f.createRoom(t, room.Persistent) for _, g := range []*porttest.Node{gateway, unreachable} { if _, err := g.Heartbeat(t.Context(), aliveTTL); err != nil { diff --git a/internal/app/realtime/observer_test.go b/internal/app/realtime/observer_test.go index 31db1c5..4ecafae 100644 --- a/internal/app/realtime/observer_test.go +++ b/internal/app/realtime/observer_test.go @@ -343,7 +343,7 @@ func TestRevocationNoticesAreBestEffort(t *testing.T) { } store := porttest.New() var logs logBuffer - _, g := New(store, store, unrevoked{}, nil, porttest.NewPreparer(), Cluster{Instance: "a", Leases: node, Liveness: node, Bus: contextBus{node}}, slog.New(slog.NewJSONHandler(&logs, nil))) + _, g := New(store, store, store, unrevoked{}, nil, porttest.NewPreparer(), Cluster{Instance: "a", Leases: node, Liveness: node, Bus: contextBus{node}}, slog.New(slog.NewJSONHandler(&logs, nil))) revokeAll := func(ctx context.Context) { g.RevokeSessions(ctx, [][]byte{[]byte("s")}) g.Revoke(ctx, "r") @@ -430,7 +430,7 @@ func TestObserverSkipsSnapshotOfEarlierRun(t *testing.T) { redis := porttest.NewRedis() node := redis.Node("a") bus := &heldEmits{Bus: node, release: make(chan struct{})} - h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler)) + h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler)) ctx, stop := context.WithCancel(context.Background()) wait, err := Start(ctx, h, g) if err != nil { diff --git a/internal/bootstrap/public.go b/internal/bootstrap/public.go index 2f0f3df..0eb31ca 100644 --- a/internal/bootstrap/public.go +++ b/internal/bootstrap/public.go @@ -76,7 +76,7 @@ func newPublic(cfg serveConfig, pool *pgxpool.Pool, rdb *goredis.Client, instanc Bus: redis.NewBus(rdb, instance, version, metrics, tracer, base.With("component", "bus")), } runtimeLogger := base.With("component", "realtime") - host, gateway := realtime.New(measure(media.NewSnapshots(store, objects, runtimeLogger), metrics), store, store, music, preparer, cluster, runtimeLogger) + host, gateway := realtime.New(measure(media.NewSnapshots(store, objects, runtimeLogger), metrics), store, store, store, music, preparer, cluster, runtimeLogger) registerRuntimeMetrics(metrics, host, gateway) rooms := lobby.New(store, leases) services := api.Services{