im_server/loadtest/im-pressure/recorder_test.go

package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"testing"
	"time"
)

func TestDeliveryRecorderSummarizesPeerGroupAndDuplicates(t *testing.T) {
	recorder := newDeliveryRecorder("run1", "a", 2)
	peer := AcceptedMessage{RunID: "run1", Kind: "p2p", ClientMid: "run1-p-b-0001-01", Topic: "p2p-one", Seq: 10,
		SenderMachine: "b", SenderIndex: 1, PeerMachine: "a", PeerIndex: 1}
	group := AcceptedMessage{RunID: "run1", Kind: "group", ClientMid: "run1-g-b-0001-01", Topic: "grp-one", Seq: 20,
		SenderMachine: "b", SenderIndex: 1}

	recorder.onData(AccountRecord{Index: 1}, testData(peer, time.Now()))
	recorder.onData(AccountRecord{Index: 1}, testData(peer, time.Now()))
	recorder.onData(AccountRecord{Index: 1}, testData(group, time.Now()))

	summary := recorder.summarize(AgentSummary{}, []AcceptedMessage{peer, group})
	if summary.ExpectedPeer != 1 || summary.ReceivedPeer != 1 {
		t.Fatalf("peer expected/received = %d/%d", summary.ExpectedPeer, summary.ReceivedPeer)
	}
	if summary.ExpectedGroup != 2 || summary.ReceivedGroup != 1 || summary.RealtimeLost != 1 {
		t.Fatalf("group expected/received/lost = %d/%d/%d", summary.ExpectedGroup, summary.ReceivedGroup, summary.RealtimeLost)
	}
	if summary.Duplicates != 1 || len(summary.Missing) != 1 || summary.Missing[0].ReceiverIndex != 2 {
		t.Fatalf("duplicates=%d missing=%+v", summary.Duplicates, summary.Missing)
	}
}

func TestDeliveryRecorderSeparatesHistoryFromRealtime(t *testing.T) {
	recorder := newDeliveryRecorder("run1", "a", 1)
	message := AcceptedMessage{RunID: "run1", Kind: "p2p", ClientMid: "run1-p-b-0001-01", Topic: "p2p-one", Seq: 10,
		SenderMachine: "b", SenderIndex: 1, PeerMachine: "a", PeerIndex: 1}
	recorder.beginHistory()
	recorder.onData(AccountRecord{Index: 1}, testData(message, time.Now()))

	summary := recorder.summarize(AgentSummary{}, []AcceptedMessage{message})
	if summary.ReceivedPeer != 0 || summary.RealtimeLost != 1 {
		t.Fatalf("history response polluted realtime result: %+v", summary)
	}
	if !recorder.historyFound(message.ClientMid) {
		t.Fatal("history message was not recorded")
	}
	if got := recorder.countRealtimeLostStored([]AcceptedMessage{message}, []string{message.ClientMid}); got != 1 {
		t.Fatalf("realtime lost but persisted=%d", got)
	}
}

func TestDeliveryRecorderUsesSharedHistoryForRealtimeLostStored(t *testing.T) {
	recorder := newDeliveryRecorder("run1", "b", 1)
	message := AcceptedMessage{RunID: "run1", Kind: "group", ClientMid: "run1-g-a-0001-01", Topic: "grp-one", Seq: 10}

	if got := recorder.countRealtimeLostStored([]AcceptedMessage{message}, []string{message.ClientMid}); got != 1 {
		t.Fatalf("shared history count=%d want=1", got)
	}
	if got := recorder.countRealtimeLostStored([]AcceptedMessage{message}, nil); got != 0 {
		t.Fatalf("empty shared history count=%d want=0", got)
	}
}

func TestDeliveryRecorderExportsObservedMessagesWithoutAck(t *testing.T) {
	recorder := newDeliveryRecorder("run1", "b", 2)
	acked := AcceptedMessage{RunID: "run1", Scenario: "group", Kind: "group", ClientMid: "run1-g-a-0001-01",
		Topic: "grp-one", Seq: 10, SenderMachine: "a", SenderIndex: 1}
	unacked := AcceptedMessage{RunID: "run1", Scenario: "group", Kind: "group", ClientMid: "run1-g-a-0002-01",
		Topic: "grp-one", Seq: 11, SenderMachine: "a", SenderIndex: 2}
	recorder.onData(AccountRecord{Index: 1}, testData(acked, time.Now()))
	recorder.onData(AccountRecord{Index: 2}, testData(unacked, time.Now()))

	messages := recorder.unackedObservedMessages([]AcceptedMessage{acked}, "group")
	if len(messages) != 1 || messages[0].ClientMid != unacked.ClientMid || messages[0].Seq != 11 ||
		messages[0].PeerMachine != "b" || messages[0].PeerIndex != 2 || messages[0].SenderIndex != 2 {
		t.Fatalf("unexpected observed ledger: %+v", messages)
	}
}

func TestDeliveryRecorderCapturesOrderingViolationDetails(t *testing.T) {
	recorder := newDeliveryRecorder("run1", "b", 1)
	newer := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-02", Topic: "p2p-one", Seq: 12}
	older := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-01", Topic: "p2p-one", Seq: 11}
	recorder.onData(AccountRecord{Index: 1}, testData(newer, time.Now()))
	recorder.onData(AccountRecord{Index: 1}, testData(older, time.Now()))

	summary := recorder.summarize(AgentSummary{}, nil)
	if summary.OutOfOrder != 1 || len(summary.OrderingViolations) != 1 {
		t.Fatalf("out_of_order=%d violations=%+v", summary.OutOfOrder, summary.OrderingViolations)
	}
	violation := summary.OrderingViolations[0]
	if violation.ReceiverMachine != "b" || violation.ReceiverIndex != 1 ||
		violation.PreviousSeq != 12 || violation.CurrentSeq != 11 || violation.ClientMid != older.ClientMid {
		t.Fatalf("unexpected violation: %+v", violation)
	}
}

func TestDeliveryRecorderSeparatesLocalAndCrossMachineLatency(t *testing.T) {
	recorder := newDeliveryRecorder("run1", "a", 1)
	local := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-01", Topic: "p2p-one", Seq: 1}
	cross := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-b-0001-01", Topic: "p2p-one", Seq: 2}
	recorder.onData(AccountRecord{Index: 1}, testData(local, time.Now().Add(-10*time.Millisecond)))
	recorder.onData(AccountRecord{Index: 1}, testData(cross, time.Now().Add(time.Second)))

	summary := recorder.summarize(AgentSummary{}, nil)
	if summary.DeliveryLatencyLocal.Count != 1 || summary.DeliveryLatencyCross.Count != 0 {
		t.Fatalf("unexpected split latency counts: local=%d cross=%d",
			summary.DeliveryLatencyLocal.Count, summary.DeliveryLatencyCross.Count)
	}
	if summary.CrossClockSkewSamples != 1 {
		t.Fatalf("cross clock skew samples=%d", summary.CrossClockSkewSamples)
	}
}

func TestDeliveryRecorderConcurrentGroupFanout(t *testing.T) {
	const users = 32
	const messages = 64
	recorder := newDeliveryRecorder("run1", "a", users)
	ledger := make([]AcceptedMessage, messages)
	for messageIndex := range ledger {
		ledger[messageIndex] = AcceptedMessage{
			RunID: "run1", Kind: "group", Topic: "grp-one", Seq: 1,
			ClientMid: fmt.Sprintf("run1-g-b-%04d-01", messageIndex+1), SenderMachine: "b",
		}
	}

	var writers sync.WaitGroup
	for receiver := 1; receiver <= users; receiver++ {
		for _, message := range ledger {
			writers.Add(1)
			go func(receiver int, message AcceptedMessage) {
				defer writers.Done()
				recorder.onData(AccountRecord{Index: receiver}, testData(message, time.Now().Add(-time.Millisecond)))
			}(receiver, message)
		}
	}
	writers.Wait()

	summary := recorder.summarize(AgentSummary{}, ledger)
	want := int64(users * messages)
	if summary.ExpectedGroup != want || summary.ReceivedGroup != want || summary.RealtimeLost != 0 {
		t.Fatalf("group expected/received/lost = %d/%d/%d, want %d/%d/0",
			summary.ExpectedGroup, summary.ReceivedGroup, summary.RealtimeLost, want, want)
	}
	if summary.Duplicates != 0 || summary.DeliveryLatency.Count != want {
		t.Fatalf("duplicates=%d latency samples=%d want=0/%d", summary.Duplicates, summary.DeliveryLatency.Count, want)
	}
}

func TestDeliveryRecorderDisconnectDiagnostics(t *testing.T) {
	recorder := newDeliveryRecorder("run1", "a", 1)
	recorder.onDisconnect(AccountRecord{Index: 1}, errors.New("websocket: close 1006 (abnormal closure): unexpected EOF"))
	recorder.onDisconnect(AccountRecord{Index: 1}, errors.New("read tcp: connection reset by peer"))

	summary := recorder.summarize(AgentSummary{}, nil)
	if summary.Disconnects != 2 || summary.DisconnectReasons["abnormal_close"] != 1 ||
		summary.DisconnectReasons["connection_reset"] != 1 || len(summary.DisconnectSamples) != 2 {
		t.Fatalf("unexpected disconnect diagnostics: count=%d reasons=%v samples=%+v",
			summary.Disconnects, summary.DisconnectReasons, summary.DisconnectSamples)
	}
	recorder.resetDisconnects()
	summary = recorder.summarize(AgentSummary{}, nil)
	if summary.Disconnects != 0 || len(summary.DisconnectReasons) != 0 || len(summary.DisconnectSamples) != 0 {
		t.Fatalf("disconnect diagnostics were not reset: %+v", summary)
	}
}

func TestBitsetCountCountsEachBitOnce(t *testing.T) {
	if got := bitsetCount([]uint64{0b1011, 0b100000}); got != 4 {
		t.Fatalf("bitset count=%d want=4", got)
	}
}

func testData(message AcceptedMessage, sentAt time.Time) dataPacket {
	content, _ := json.Marshal(map[string]any{"run_id": message.RunID, "scenario": message.Scenario,
		"sender_machine": message.SenderMachine, "sender_index": message.SenderIndex,
		"sent_at_unix_nano": sentAt.UnixNano()})
	return dataPacket{Topic: message.Topic, Seq: message.Seq,
		Head: map[string]any{"x-im-client-mid": message.ClientMid}, Content: content}
}