local_network.go

package paxos

import (
	"context"
	"math/rand/v2"
	"time"
)

type network interface {
	Send(ctx context.Context, from, to int, msg Message)
}

type messageReceiver interface {
	receive(from int, msg Message)
}

type localNetwork struct {
	receivers map[int]messageReceiver
	faults    *FaultInjector
	maxJitter time.Duration
}

func newLocalNetwork(faults *FaultInjector, maxJitter time.Duration) *localNetwork {
	return &localNetwork{
		receivers: make(map[int]messageReceiver),
		faults:    faults,
		maxJitter: maxJitter,
	}
}

func (n *localNetwork) register(id int, receiver messageReceiver) {
	n.receivers[id] = receiver
}

func (n *localNetwork) deliver(ctx context.Context, from, to int, msg Message) {
	select {
	case <-ctx.Done():
		return
	default:
	}

	if receiver, ok := n.receivers[to]; ok {
		receiver.receive(from, msg)
	}
}

func (n *localNetwork) deliverAfter(ctx context.Context, delay time.Duration, from, to int, msg Message) {
	if !n.waitForDelay(ctx, delay) {
		return
	}
	n.deliver(ctx, from, to, msg)
}

func (n *localNetwork) Send(ctx context.Context, from, to int, msg Message) {
	plan := n.faults.PlanDelivery(from, to, msg)

	switch plan.action {
	case forward:
		go n.deliverAfter(ctx, n.randomJitter(), from, to, msg)
	case delay:
		go n.deliverAfter(ctx, plan.delay+n.randomJitter(), from, to, msg)
	case duplicate:
		go n.deliverAfter(ctx, n.randomJitter(), from, to, msg)
		go n.deliverAfter(ctx, n.randomJitter(), from, to, msg)
	case drop:
		return
	}
}

func (n *localNetwork) randomJitter() time.Duration {
	if n.maxJitter <= 0 {
		return 0
	}
	return rand.N(n.maxJitter)
}

func (n *localNetwork) waitForDelay(ctx context.Context, delay time.Duration) bool {
	if delay <= 0 {
		return true
	}

	timer := time.NewTimer(delay)
	defer timer.Stop()

	select {
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}