fix(all): stop activity refresh loops with changed mission notifications and trace event requests

go/internal/server/eventtasks/notify_changes.go

@@ -0,0 +1,114 @@
+package eventtasks
+
+import (
+	"bd2server/internal/server/wire"
+	"fmt"
+	"sort"
+)
+
+// The baseline belongs to the current account transaction, not a client session.
+// Reads and retries therefore do not replay historical progress as notifications.
+type missionNoticeKey struct {
+	Schedule           string
+	Event, Group, Task uint64
+}
+
+// visibleMissionValues is a read-only view. In particular, it must not call
+// mission(), which initializes and resets persisted progress as a side effect.
+func (s *Service) visibleMissionValues() map[missionNoticeKey]uint64 {
+	out := map[missionNoticeKey]uint64{}
+	for _, v := range s.taskSchedules() {
+		if v.Type != 4 || !s.active(v) {
+			continue
+		}
+		group := s.design.MissionGroups[v.ID]
+		period := ""
+		if group.Type == 1 {
+			period = s.day()
+		}
+		if group.Type == 2 {
+			y, w := s.now().UTC().ISOWeek()
+			period = fmt.Sprintf("%d-%d", y, w)
+		}
+		for _, t := range s.design.Missions {
+			if !s.availableTask(v, t) {
+				continue
+			}
+			k := missionNoticeKey{scheduleKey(v), v.ID, t.Group, t.ID}
+			value := uint64(0)
+			if m := s.state.Missions[k.Schedule+"/"+key(t.ID)]; m != nil && m.Period == period {
+				value = m.Value
+			}
+			out[k] = value
+		}
+	}
+	return out
+}
+
+func (s *Service) notifyMissionChanges() ([]byte, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.beforeMissions == nil {
+		return nil, nil
+	}
+	current := s.visibleMissionValues()
+	var keys []missionNoticeKey
+	for k, value := range current {
+		if value != s.beforeMissions[k] {
+			keys = append(keys, k)
+		}
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		a, b := keys[i], keys[j]
+		if a.Event != b.Event {
+			return a.Event < b.Event
+		}
+		if a.Group != b.Group {
+			return a.Group < b.Group
+		}
+		if a.Task != b.Task {
+			return a.Task < b.Task
+		}
+		return a.Schedule < b.Schedule
+	})
+	var out, groupMap, rows []byte
+	var eventID, groupID uint64
+	flushGroup := func() {
+		if len(rows) == 0 {
+			return
+		}
+		entry := wire.AppendVarint(nil, 1, groupID)
+		entry = wire.AppendBytes(entry, 2, rows)
+		groupMap = wire.AppendBytes(groupMap, 1, entry)
+		rows = nil
+	}
+	flushEvent := func() {
+		flushGroup()
+		if len(groupMap) == 0 {
+			return
+		}
+		entry := wire.AppendVarint(nil, 1, eventID)
+		entry = wire.AppendBytes(entry, 2, groupMap)
+		out = wire.AppendBytes(out, 4, entry)
+		groupMap = nil
+	}
+	for _, k := range keys {
+		if k.Event != eventID {
+			flushEvent()
+			eventID = k.Event
+			groupID = k.Group
+		}
+		if k.Group != groupID {
+			flushGroup()
+			groupID = k.Group
+		}
+		entry := wire.AppendVarint(nil, 1, k.Task)
+		entry = wire.AppendVarint(entry, 2, current[k])
+		rows = wire.AppendBytes(rows, 1, entry)
+	}
+	flushEvent()
+	// Consume only this request's baseline; the next BeforeDispatch takes a fresh
+	// snapshot even when a transaction rolls back or the session changes.
+	s.beforeMissions = nil
+	return out, nil
+}

go/internal/server/eventtasks/notify_changes_test.go

@@ -0,0 +1,183 @@
+package eventtasks
+
+import (
+	"bd2server/internal/server/events"
+	"bd2server/internal/server/gamedata"
+	"bd2server/internal/server/wire"
+	"bd2server/internal/server/world"
+	"bytes"
+	"encoding/json"
+	"testing"
+)
+
+func notice(t *testing.T, s *Service, seq uint64) []byte {
+	t.Helper()
+	b, err := s.AfterDispatch("/MiniGameRouletteInfo", req(seq), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return b
+}
+
+func TestRouletteReadsDoNotReplayHistoricalMissionNotifications(t *testing.T) {
+	s, _, _ := setup(t)
+	if err := s.RecordEvent(2, 0, 1, nil); err != nil {
+		t.Fatal(err)
+	}
+	for _, session := range []string{"test", "new-session"} {
+		s.BeginSession(session)
+		for seq := uint64(1); seq <= 3; seq++ {
+			if err := s.BeforeDispatch("/MiniGameRouletteInfo", req(seq)); err != nil {
+				t.Fatal(err)
+			}
+			if got := notice(t, s, seq); len(got) != 0 {
+				t.Fatalf("read notified old mission: %x", got)
+			}
+		}
+	}
+}
+
+func TestMissionNotificationContainsOnlyChangesAndAbsoluteZero(t *testing.T) {
+	s, _, _ := setup(t)
+	before, _ := json.Marshal(s.state)
+	if err := s.BeforeDispatch("/MiniGameRouletteInfo", req(1)); err != nil {
+		t.Fatal(err)
+	}
+	after, _ := json.Marshal(s.state)
+	if !bytes.Equal(before, after) {
+		t.Fatal("baseline initialized persistent missions")
+	}
+	if err := s.RecordEvent(2, 0, 1, nil); err != nil {
+		t.Fatal(err)
+	}
+	first := notice(t, s, 1)
+	if len(first) == 0 {
+		t.Fatal("actual progress did not notify")
+	}
+	if len(notice(t, s, 1)) != 0 {
+		t.Fatal("same request notified twice")
+	}
+	if err := s.BeforeDispatch("/MiniGameRouletteInfo", req(2)); err != nil {
+		t.Fatal(err)
+	}
+	if len(notice(t, s, 2)) != 0 {
+		t.Fatal("subsequent read notified")
+	}
+	if err := s.BeforeDispatch("/reset", req(3)); err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range s.state.Missions {
+		m.Value = 0
+	}
+	zero := notice(t, s, 3)
+	if len(zero) == 0 {
+		t.Fatal("zero transition was omitted")
+	}
+	var value uint64 = 99
+	_ = wire.Walk(zero, func(event wire.Field) error {
+		groupMap, _, _ := wire.Bytes(event.Value, 2)
+		return wire.Walk(groupMap, func(group wire.Field) error {
+			rows, _, _ := wire.Bytes(group.Value, 2)
+			return wire.Walk(rows, func(row wire.Field) error { value = scalar(row.Value, 2); return nil })
+		})
+	})
+	if value != 0 {
+		t.Fatalf("notification is not absolute zero: %d", value)
+	}
+}
+
+type noticeProvider struct{ count uint64 }
+
+func (p *noticeProvider) Snapshot() (world.GameplayAchievementSnapshot, error) {
+	return world.GameplayAchievementSnapshot{Items: map[[2]uint64]uint64{{5, 8}: p.count}}, nil
+}
+func (*noticeProvider) Events(string, []byte, []byte, world.GameplayAchievementSnapshot, world.GameplayAchievementSnapshot) ([]world.GameplayAchievementEvent, error) {
+	return nil, nil
+}
+
+func TestObserverBatchRollbackRetryKeepsMissionDelta(t *testing.T) {
+	s, _, store := setup(t)
+	task := s.design.Missions[10]
+	task.Type = 32
+	s.design.Missions[10] = task
+	p := &noticeProvider{count: 1}
+	s.AttachGameplayProvider(p)
+	if err := s.save(); err != nil {
+		t.Fatal(err)
+	}
+	checkpoint, err := store.Load("eventtasks")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = s.BeforeDispatch("/batch", req(1)); err != nil {
+		t.Fatal(err)
+	}
+	p.count = 2
+	first := notice(t, s, 1)
+	if len(first) == 0 {
+		t.Fatal("observer item gain did not notify")
+	}
+	// Account transaction rollback restores the domain snapshot and inventory.
+	if err = store.Save("eventtasks", checkpoint); err != nil {
+		t.Fatal(err)
+	}
+	s.state = snapshot{}
+	if err = json.Unmarshal(checkpoint, &s.state); err != nil {
+		t.Fatal(err)
+	}
+	p.count = 1
+	if err = s.BeforeDispatch("/batch", req(1)); err != nil {
+		t.Fatal(err)
+	}
+	p.count = 2
+	retry := notice(t, s, 1)
+	if !bytes.Equal(first, retry) {
+		t.Fatalf("rolled-back retry lost delta: %x / %x", first, retry)
+	}
+	if err = s.BeforeDispatch("/batch", req(1)); err != nil {
+		t.Fatal(err)
+	}
+	if len(notice(t, s, 1)) != 0 {
+		t.Fatal("committed retry repeated delta")
+	}
+}
+
+func TestMissionDeltaSeparatesPermanentGroupsAndOmitsEmptyGroups(t *testing.T) {
+	s, _, _ := setup(t)
+	r := events.NewRegistry()
+	if err := r.Replace([]events.Schedule{{UID: 0, Type: 4, ID: 7, Start: 1, End: 9999999999999}, {UID: 0, Type: 4, ID: 8, Start: 1, End: 9999999999999}}); err != nil {
+		t.Fatal(err)
+	}
+	s.registry = r
+	s.design.MissionGroups[8] = gamedata.EventMissionGroup{ID: 8, Groups: []uint64{11}}
+	s.design.Missions[12] = gamedata.EventTask{ID: 12, Group: 11, Type: 99, Target: 2}
+	if err := s.RecordEvent(99, 0, 1, nil); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.BeforeDispatch("/play", req(1)); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RecordEvent(2, 0, 1, nil); err != nil {
+		t.Fatal(err)
+	}
+	got := notice(t, s, 1)
+	count := 0
+	if err := wire.Walk(got, func(f wire.Field) error {
+		count++
+		if f.Number != 4 || scalar(f.Value, 1) != 7 {
+			t.Fatalf("unchanged group emitted: %x", f.Value)
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if count != 1 {
+		t.Fatalf("wanted changed group only, got %d", count)
+	}
+	if err := s.BeforeDispatch("/read", req(2)); err != nil {
+		t.Fatal(err)
+	}
+	if len(notice(t, s, 2)) != 0 {
+		t.Fatal("empty event/group emitted")
+	}
+}

go/internal/server/eventtasks/observer.go

@@ -8,6 +8,9 @@ import (
 
 func (s *Service) BeginSession(id string) { s.SetSession(id) }
 func (s *Service) BeforeDispatch(string, []byte) error {
+	s.mu.Lock()
+	s.beforeMissions = s.visibleMissionValues()
+	s.mu.Unlock()
 	if s.provider == nil {
 		return nil
 	}
@@ -72,7 +75,7 @@ func (s *Service) AfterDispatch(path string, request, response []byte) ([]byte,
 			}
 		}
 	}
-	return s.Notify()
+	return s.notifyMissionChanges()
 }
 
 func (s *Service) CompleteSingleTargetEvent(condition uint64, unlocked func(uint64, uint64) bool) error {

go/internal/server/eventtasks/service.go

@@ -57,6 +57,7 @@ type snapshot struct {
 type Service struct {
 	provider          world.GameplayAchievementProvider
 	before            world.GameplayAchievementSnapshot
+	beforeMissions    map[missionNoticeKey]uint64
 	mu                sync.Mutex
 	store             stateio.Store
 	design            *gamedata.EventTasksDesign