tessera/cmd/datad/main.go

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/google/uuid"
	"gitlab.michelsen.id/phillmichelsen/tessera/pkg/data"
	"gitlab.michelsen.id/phillmichelsen/tessera/pkg/data/routing"
)

type SeqPayload struct {
	Seq uint64
}

type streamStats struct {
	sent       uint64
	observed   uint64
	missed     uint64
	lastSeen   uint64
	lastReport time.Time
}

func main() {
	ctx := context.Background()

	// ---- Knobs ----
	N := 10
	duration := 5 * time.Second
	totalTargetPerSec := 5_000 // total across all streams
	// ----------------

	rt := routing.NewInprocRouter()

	senders := make([]data.Sender, N)
	receivers := make([]data.Receiver, N)

	for i := range N {
		st, err := rt.OpenStream(data.StreamID(uuid.New()))
		if err != nil {
			panic(err)
		}
		senders[i] = st.Sender()
		receivers[i] = st.Receiver()
	}

	perStreamTarget := totalTargetPerSec / N
	if perStreamTarget == 0 {
		perStreamTarget = 1
	}

	fmt.Printf("N=%d duration=%s totalTarget=%d/s perStreamTarget=%d/s\n",
		N, duration, totalTargetPerSec, perStreamTarget)

	stopAt := time.Now().Add(duration)

	stats := make([]streamStats, N)

	var wg sync.WaitGroup
	wg.Add(N + 1)

	// Publisher: per-stream sender sequence in envelope payload.
	go func() {
		defer wg.Done()

		tick := time.NewTicker(1 * time.Millisecond)
		defer tick.Stop()

		perTick := perStreamTarget / 1000
		rem := perStreamTarget % 1000
		remAcc := make([]int, N)

		seq := make([]uint64, N)

		for time.Now().Before(stopAt) {
			<-tick.C

			for i := range N {
				n := int(perTick)
				remAcc[i] += rem
				if remAcc[i] >= 1000 {
					n++
					remAcc[i] -= 1000
				}

				for j := 0; j < n; j++ {
					seq[i]++

					env := data.Envelope{
						Payload: SeqPayload{Seq: seq[i]},
					}
					_ = senders[i].Send(ctx, env)
					stats[i].sent++
				}
			}
		}
	}()

	// Consumers: detect missed sender sequence numbers.
	for i := range N {
		idx := i
		rx := receivers[i]

		go func() {
			defer wg.Done()

			for time.Now().Before(stopAt) {
				env, ok, err := rx.TryReceive()
				if err != nil {
					return
				}
				if !ok {
					continue
				}

				p, ok := env.Payload.(SeqPayload)
				if !ok {
					// If your Payload is pointer/interface-heavy, adjust accordingly.
					continue
				}

				stats[idx].observed++

				if stats[idx].lastSeen == 0 {
					stats[idx].lastSeen = p.Seq
					continue
				}

				if p.Seq > stats[idx].lastSeen+1 {
					stats[idx].missed += (p.Seq - stats[idx].lastSeen - 1)
				}
				stats[idx].lastSeen = p.Seq
			}
		}()
	}

	wg.Wait()

	var totalSent, totalObs, totalMiss uint64
	minDrop, maxDrop := 100.0, 0.0

	for i := range N {
		totalSent += stats[i].sent
		totalObs += stats[i].observed
		totalMiss += stats[i].missed

		den := stats[i].observed + stats[i].missed
		dropPct := 0.0
		if den > 0 {
			dropPct = 100.0 * float64(stats[i].missed) / float64(den)
		}
		if dropPct < minDrop {
			minDrop = dropPct
		}
		if dropPct > maxDrop {
			maxDrop = dropPct
		}

		fmt.Printf("stream[%02d] sent=%6d observed=%6d missed=%6d lastSeen=%6d drop=%5.2f%%\n",
			i, stats[i].sent, stats[i].observed, stats[i].missed, stats[i].lastSeen, dropPct)
	}

	totalDen := totalObs + totalMiss
	totalDrop := 0.0
	if totalDen > 0 {
		totalDrop = 100.0 * float64(totalMiss) / float64(totalDen)
	}

	fmt.Printf("\nTOTAL sent=%d observed=%d missed=%d drop=%.2f%% (min=%.2f%% max=%.2f%%)\n",
		totalSent, totalObs, totalMiss, totalDrop, minDrop, maxDrop)
}