Mesh frames#

use pamoja_mesh::{Frame, SeenCache, BROADCAST};

// A river gauge floods a level reading to every node in range. The header is fixed
// and big-endian: version, source, destination, sequence id, hop limit, then the
// payload and a checksum over everything but the hop limit.
const RIVER_GAUGE: u32 = 305_419_896;
let reading = Frame::broadcast(RIVER_GAUGE, 1, b"level=high")?;
let to = if reading.dst() == BROADCAST {
    "every node in range"
} else {
    "one node"
};
println!(
    "sent      {} bytes to {to}, hop limit {}",
    reading.as_bytes().len(),
    reading.hop_limit()
);

// A neighbor hears it. Every node in range rebroadcasts, so the same packet arrives
// several times over; the source and sequence id decide which copy is the first.
let received = Frame::parse(reading.as_bytes())?;
println!("payload   {}", String::from_utf8_lossy(received.payload()));
let mut seen: SeenCache<64> = SeenCache::new();
let first = seen.record(received.dedup_key());
let again = seen.record(received.dedup_key());
if first && !again {
    println!("dedup     the first copy is relayed, and the second is dropped");
}

// Relaying spends one hop. The checksum skips the hop-limit byte, so a relay forwards
// the frame without recomputing it and the check stays end to end.
let forwarded = received.relayed().expect("hops remain");
let onward = Frame::parse(forwarded.as_bytes())?;
println!(
    "relayed   hop limit {}, and the checksum still holds: {}",
    forwarded.hop_limit(),
    String::from_utf8_lossy(onward.payload())
);

// A frame that has run out of hops is not relayed again, which is what ends the flood.
if received.with_hop_limit(0).relayed().is_none() {
    println!("spent     at hop limit 0 the frame goes no further");
}

// A payload byte the air mangled fails the checksum rather than reaching the
// application as a plausible reading. The header is a fixed width, so the first byte
// past it is the first byte of the reading itself.
let mut mangled = reading.as_bytes().to_vec();
mangled[Frame::HEADER_LEN] ^= 0xFF;
match Frame::parse(&mangled) {
    Ok(_) => println!("a mangled frame was accepted, which should never happen"),
    Err(error) => println!("mangled   rejected: {error}"),
}

// Six nodes down a river valley, each in range of its neighbors only. The gauge at the
// top, node 0, floods a reading, and every node that hears a packet for the first time
// takes it and relays it while hops remain. Copies that come back up the valley are
// echoes, which the memory of each node drops.
let nodes = 6;
let mut memory = vec![SeenCache::<64>::new(); nodes];
let flooded = Frame::broadcast(0, 1, b"level=high")?;
memory[0].record(flooded.dedup_key());
let mut on_the_air = std::collections::VecDeque::from([(0usize, flooded)]);
let (mut delivered, mut relays, mut echoes, mut farthest) = (0, 0, 0, 0);
while let Some((from, frame)) = on_the_air.pop_front() {
    for node in [from.checked_sub(1), Some(from + 1)].into_iter().flatten() {
        if node >= nodes {
            continue;
        }
        let heard = Frame::parse(frame.as_bytes())?;
        if !memory[node].record(heard.dedup_key()) {
            echoes += 1;
            continue;
        }
        delivered += 1;
        farthest = farthest.max(node);
        if let Some(onward) = heard.relayed() {
            relays += 1;
            on_the_air.push_back((node, onward));
        }
    }
}
println!(
    "flood     {delivered} nodes took the reading, {relays} relayed it, {echoes} echoes were dropped"
);
println!(
    "reach     node {farthest} was the farthest, {farthest} hops out, and node {} never heard it",
    nodes - 1
);

// A node whose memory holds two packets hears the reading, then two packets from
// other nodes, then a late copy of the reading by a longer path. It has forgotten the
// reading by then and relays it again; on a busy mesh that repeats without end.
let mut small: SeenCache<2> = SeenCache::new();
let rain = Frame::broadcast(7, 1, b"rain=4mm")?;
let wind = Frame::broadcast(8, 1, b"wind=12")?;
small.record(flooded.dedup_key());
small.record(rain.dedup_key());
small.record(wind.dedup_key());
if small.record(flooded.dedup_key()) {
    println!("forgot    a memory of two packets relays the late copy again");
}

// A payload one byte past what a frame carries is refused before it is built. A frame
// is sized to the 250 bytes ESP-NOW carries, less the header and the checksum.
match Frame::broadcast(0, 2, &[0u8; Frame::MAX_PAYLOAD + 1]) {
    Ok(_) => println!("an oversized payload was framed, which should never happen"),
    Err(error) => println!(
        "too long  {} bytes refused: {error}",
        Frame::MAX_PAYLOAD + 1
    ),
}

import { BROADCAST, HEADER_LEN, SeenPackets, broadcast, parse, relayed } from '@pamoja/mesh'

// A river gauge floods a level reading to every node in range. The header is fixed and
// big-endian: version, source, destination, sequence id, hop limit, then the payload and
// a checksum over everything but the hop limit.
const RIVER_GAUGE = 305419896
const reading = broadcast(RIVER_GAUGE, 1, Buffer.from('level=high'))
const to = reading.dst === BROADCAST ? 'every node in range' : 'one node'
console.log(`sent      ${reading.bytes.length} bytes to ${to}, hop limit ${reading.hopLimit}`)

// A neighbor hears it. Every node in range rebroadcasts, so the same packet arrives
// several times over; the source and sequence id decide which copy is the first.
const received = parse(reading.bytes)
console.log(`payload   ${received.payload.toString()}`)
const seen = new SeenPackets(64)
const first = seen.record(received.src, received.id)
const again = seen.record(received.src, received.id)
if (first && !again) {
  console.log('dedup     the first copy is relayed, and the second is dropped')
}

// Relaying spends one hop. The checksum skips the hop-limit byte, so a relay forwards the
// frame without recomputing it and the check stays end to end.
const forwarded = relayed(received.bytes)!
const onward = parse(forwarded.bytes)
console.log(
  `relayed   hop limit ${forwarded.hopLimit}, and the checksum still holds: ${onward.payload.toString()}`,
)

// A frame that has run out of hops is not relayed again, which is what ends the flood.
if (relayed(broadcast(RIVER_GAUGE, 1, Buffer.from('level=high'), 0).bytes) === null) {
  console.log('spent     at hop limit 0 the frame goes no further')
}

// A payload byte the air mangled fails the checksum rather than reaching the application
// as a plausible reading. The header is a fixed width, so the first byte past it is the
// first byte of the reading itself.
const mangled = Buffer.from(reading.bytes)
mangled[HEADER_LEN] ^= 0xff
try {
  parse(mangled)
  console.log('a mangled frame was accepted, which should never happen')
} catch (error) {
  console.log(`mangled   rejected: ${(error as Error).message}`)
}

import { DEFAULT_HOP_LIMIT, MAX_PAYLOAD } from '@pamoja/mesh'

// Six nodes down a river valley, each in range of its neighbors only. The gauge at the top,
// node 0, floods a reading, and every node that hears a packet for the first time takes it
// and relays it while hops remain. Copies that come back up the valley are echoes, which the
// memory of each node drops.
const nodes = 6
const memory = Array.from({ length: nodes }, () => new SeenPackets(64))
const flooded = broadcast(0, 1, Buffer.from('level=high'))
memory[0].record(flooded.src, flooded.id)
const onTheAir = [{ from: 0, bytes: flooded.bytes }]
let delivered = 0
let relays = 0
let echoes = 0
let farthest = 0
for (let sent = onTheAir.shift(); sent; sent = onTheAir.shift()) {
  for (const node of [sent.from - 1, sent.from + 1]) {
    if (node < 0 || node >= nodes) {
      continue
    }
    const heard = parse(sent.bytes)
    if (!memory[node].record(heard.src, heard.id)) {
      echoes += 1
      continue
    }
    delivered += 1
    farthest = Math.max(farthest, node)
    const onwardFrame = relayed(heard.bytes)
    if (onwardFrame) {
      relays += 1
      onTheAir.push({ from: node, bytes: onwardFrame.bytes })
    }
  }
}
console.log(`flood     ${delivered} nodes took the reading, ${relays} relayed it, ${echoes} echoes were dropped`)
console.log(
  `reach     node ${farthest} was the farthest, ${farthest} hops out, and node ${nodes - 1} never heard it`,
)

// A node whose memory holds two packets hears the reading, then two packets from other
// nodes, then a late copy of the reading by a longer path. It has forgotten the reading by
// then and relays it again; on a busy mesh that repeats without end.
const small = new SeenPackets(2)
const rain = broadcast(7, 1, Buffer.from('rain=4mm'))
const wind = broadcast(8, 1, Buffer.from('wind=12'))
small.record(flooded.src, flooded.id)
small.record(rain.src, rain.id)
small.record(wind.src, wind.id)
if (small.record(flooded.src, flooded.id)) {
  console.log('forgot    a memory of two packets relays the late copy again')
}

// A payload one byte past what a frame carries is refused before it is built. A frame is
// sized to the 250 bytes ESP-NOW carries, less the header and the checksum.
try {
  broadcast(0, 2, Buffer.alloc(MAX_PAYLOAD + 1))
  console.log('an oversized payload was framed, which should never happen')
} catch (error) {
  console.log(`too long  ${MAX_PAYLOAD + 1} bytes refused: ${(error as Error).message}`)
}

from pamoja.core import PamojaError
from pamoja.mesh import BROADCAST, HEADER_LEN, SeenPackets, broadcast, parse, relayed

# A river gauge floods a level reading to every node in range. The header is fixed and
# big-endian: version, source, destination, sequence id, hop limit, then the payload and a
# checksum over everything but the hop limit.
RIVER_GAUGE = 305419896
reading = broadcast(RIVER_GAUGE, 1, b"level=high")
to = "every node in range" if reading.dst == BROADCAST else "one node"
print(f"sent      {len(reading.bytes)} bytes to {to}, hop limit {reading.hop_limit}")

# A neighbor hears it. Every node in range rebroadcasts, so the same packet arrives several
# times over; the source and sequence id decide which copy is the first.
received = parse(reading.bytes)
print(f"payload   {received.payload.decode()}")
seen = SeenPackets(64)
first = seen.record(received.src, received.id)
again = seen.record(received.src, received.id)
if first and not again:
    print("dedup     the first copy is relayed, and the second is dropped")

# Relaying spends one hop. The checksum skips the hop-limit byte, so a relay forwards the
# frame without recomputing it and the check stays end to end.
forwarded = relayed(received.bytes)
onward = parse(forwarded.bytes)
print(
    f"relayed   hop limit {forwarded.hop_limit}, and the checksum still holds: "
    f"{onward.payload.decode()}"
)

# A frame that has run out of hops is not relayed again, which is what ends the flood.
if relayed(broadcast(RIVER_GAUGE, 1, b"level=high", hop_limit=0).bytes) is None:
    print("spent     at hop limit 0 the frame goes no further")

# A payload byte the air mangled fails the checksum rather than reaching the application as a
# plausible reading. The header is a fixed width, so the first byte past it is the first byte
# of the reading itself.
mangled = bytearray(reading.bytes)
mangled[HEADER_LEN] ^= 0xFF
try:
    parse(bytes(mangled))
    print("a mangled frame was accepted, which should never happen")
except PamojaError as error:
    print(f"mangled   rejected: {error}")

from collections import deque

from pamoja.mesh import DEFAULT_HOP_LIMIT, MAX_PAYLOAD

# Six nodes down a river valley, each in range of its neighbors only. The gauge at the top,
# node 0, floods a reading, and every node that hears a packet for the first time takes it
# and relays it while hops remain. Copies that come back up the valley are echoes, which the
# memory of each node drops.
nodes = 6
memory = [SeenPackets(64) for _ in range(nodes)]
flooded = broadcast(0, 1, b"level=high")
memory[0].record(flooded.src, flooded.id)
on_the_air = deque([(0, flooded.bytes)])
delivered = relays = echoes = farthest = 0
while on_the_air:
    sender, sent = on_the_air.popleft()
    for node in (sender - 1, sender + 1):
        if not 0 <= node < nodes:
            continue
        heard = parse(sent)
        if not memory[node].record(heard.src, heard.id):
            echoes += 1
            continue
        delivered += 1
        farthest = max(farthest, node)
        onward_frame = relayed(heard.bytes)
        if onward_frame is not None:
            relays += 1
            on_the_air.append((node, onward_frame.bytes))
print(f"flood     {delivered} nodes took the reading, {relays} relayed it, {echoes} echoes were dropped")
print(
    f"reach     node {farthest} was the farthest, {farthest} hops out, "
    f"and node {nodes - 1} never heard it"
)

# A node whose memory holds two packets hears the reading, then two packets from other nodes,
# then a late copy of the reading by a longer path. It has forgotten the reading by then and
# relays it again; on a busy mesh that repeats without end.
small = SeenPackets(2)
rain = broadcast(7, 1, b"rain=4mm")
wind = broadcast(8, 1, b"wind=12")
small.record(flooded.src, flooded.id)
small.record(rain.src, rain.id)
small.record(wind.src, wind.id)
if small.record(flooded.src, flooded.id):
    print("forgot    a memory of two packets relays the late copy again")

# A payload one byte past what a frame carries is refused before it is built. A frame is
# sized to the 250 bytes ESP-NOW carries, less the header and the checksum.
try:
    broadcast(0, 2, bytes(MAX_PAYLOAD + 1))
    print("an oversized payload was framed, which should never happen")
except PamojaError as error:
    print(f"too long  {MAX_PAYLOAD + 1} bytes refused: {error}")

// A river gauge floods a level reading to every node in range. The header is fixed
// and big-endian: version, source, destination, sequence id, hop limit, then the
// payload and a checksum over everything but the hop limit.
const uint RiverGauge = 305_419_896;
MeshFrame reading = Mesh.BroadcastFrame(RiverGauge, 1, "level=high"u8);
string to = reading.Dst == Mesh.Broadcast ? "every node in range" : "one node";
Console.WriteLine($"sent      {reading.Bytes.Length} bytes to {to}, hop limit {reading.HopLimit}");

// A neighbor hears it. Every node in range rebroadcasts, so the same packet
// arrives several times over; the source and sequence id decide which copy is
// the first.
MeshFrame received = Mesh.Parse(reading.Bytes);
Console.WriteLine($"payload   {Encoding.UTF8.GetString(received.Payload)}");
using SeenPackets seen = new(64);
bool first = seen.Record(received.Src, received.Id);
bool again = seen.Record(received.Src, received.Id);
if (first && !again)
{
    Console.WriteLine("dedup     the first copy is relayed, and the second is dropped");
}

// Relaying spends one hop. The checksum skips the hop-limit byte, so a relay
// forwards the frame without recomputing it and the check stays end to end.
MeshFrame forwarded = Mesh.Relayed(received.Bytes)!;
MeshFrame onward = Mesh.Parse(forwarded.Bytes);
Console.WriteLine(
    $"relayed   hop limit {forwarded.HopLimit}, and the checksum still holds: " +
    Encoding.UTF8.GetString(onward.Payload));

// A frame that has run out of hops is not relayed again, which is what ends the
// flood.
if (Mesh.Relayed(Mesh.BroadcastFrame(RiverGauge, 1, "level=high"u8, hopLimit: 0).Bytes) is null)
{
    Console.WriteLine("spent     at hop limit 0 the frame goes no further");
}

// A payload byte the air mangled fails the checksum rather than reaching the
// application as a plausible reading. The header is a fixed width, so the first
// byte past it is the first byte of the reading itself.
byte[] mangled = reading.Bytes.ToArray();
mangled[Mesh.HeaderLen] ^= 0xFF;
try
{
    Mesh.Parse(mangled);
    Console.WriteLine("a mangled frame was accepted, which should never happen");
}
catch (PamojaException error)
{
    Console.WriteLine($"mangled   rejected: {error.Message}");
}

// Six nodes down a river valley, each in range of its neighbors only. The gauge at
// the top, node 0, floods a reading, and every node that hears a packet for the
// first time takes it and relays it while hops remain. Copies that come back up the
// valley are echoes, which the memory of each node drops.
const int Nodes = 6;
SeenPackets[] memory = Enumerable.Range(0, Nodes).Select(_ => new SeenPackets(64)).ToArray();
MeshFrame flooded = Mesh.BroadcastFrame(0, 1, "level=high"u8);
memory[0].Record(flooded.Src, flooded.Id);
var onTheAir = new Queue<(int From, byte[] Bytes)>();
onTheAir.Enqueue((0, flooded.Bytes));
int delivered = 0;
int relays = 0;
int echoes = 0;
int farthest = 0;
while (onTheAir.TryDequeue(out var sent))
{
    foreach (int node in new[] { sent.From - 1, sent.From + 1 })
    {
        if (node < 0 || node >= Nodes)
        {
            continue;
        }

        MeshFrame heard = Mesh.Parse(sent.Bytes);
        if (!memory[node].Record(heard.Src, heard.Id))
        {
            echoes++;
            continue;
        }

        delivered++;
        farthest = Math.Max(farthest, node);
        if (Mesh.Relayed(heard.Bytes) is { } onwardFrame)
        {
            relays++;
            onTheAir.Enqueue((node, onwardFrame.Bytes));
        }
    }
}

foreach (SeenPackets cache in memory)
{
    cache.Dispose();
}

Console.WriteLine($"flood     {delivered} nodes took the reading, {relays} relayed it, {echoes} echoes were dropped");
Console.WriteLine(
    $"reach     node {farthest} was the farthest, {farthest} hops out, and node {Nodes - 1} never heard it");

// A node whose memory holds two packets hears the reading, then two packets from
// other nodes, then a late copy of the reading by a longer path. It has forgotten
// the reading by then and relays it again; on a busy mesh that repeats without end.
using SeenPackets small = new(2);
MeshFrame rain = Mesh.BroadcastFrame(7, 1, "rain=4mm"u8);
MeshFrame wind = Mesh.BroadcastFrame(8, 1, "wind=12"u8);
small.Record(flooded.Src, flooded.Id);
small.Record(rain.Src, rain.Id);
small.Record(wind.Src, wind.Id);
if (small.Record(flooded.Src, flooded.Id))
{
    Console.WriteLine("forgot    a memory of two packets relays the late copy again");
}

// A payload one byte past what a frame carries is refused before it is built. A
// frame is sized to the 250 bytes ESP-NOW carries, less the header and the checksum.
try
{
    Mesh.BroadcastFrame(0, 2, new byte[Mesh.MaxPayload + 1]);
    Console.WriteLine("an oversized payload was framed, which should never happen");
}
catch (PamojaException error)
{
    Console.WriteLine($"too long  {Mesh.MaxPayload + 1} bytes refused: {error.Message}");
}