[WIP] reliable UDP module with connection management and flow control

cmd/astrald/mods.go

@@ -22,5 +22,6 @@ import (
 	_ "github.com/cryptopunkscc/astrald/mod/shell/src"
 	_ "github.com/cryptopunkscc/astrald/mod/tcp/src"
 	_ "github.com/cryptopunkscc/astrald/mod/tor/src"
+	_ "github.com/cryptopunkscc/astrald/mod/udp/src"
 	_ "github.com/cryptopunkscc/astrald/mod/user/src"
 )

go.mod

@@ -31,6 +31,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.17 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	github.com/smallnest/ringbuffer v0.0.0-20250317021400-0da97b586904 // indirect
 	golang.org/x/sys v0.32.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	modernc.org/libc v1.24.1 // indirect

go.sum

@@ -43,6 +43,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
+github.com/smallnest/ringbuffer v0.0.0-20250317021400-0da97b586904 h1:OoG1xZV7CXnP2/Udl1ybEgTEds9XXA3NHWg+OR3c/a8=
+github.com/smallnest/ringbuffer v0.0.0-20250317021400-0da97b586904/go.mod h1:tAG61zBM1DYRaGIPloumExGvScf08oHuo0kFoOqdbT0=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/wailsapp/mimetype v1.4.1 h1:pQN9ycO7uo4vsUUuPeHEYoUkLVkaRntMnHJxVwYhwHs=

mod/nodes/src/op_streams.go

@@ -1,10 +1,11 @@
 package nodes
 
 import (
+	"slices"
+
 	"github.com/cryptopunkscc/astrald/astral"
 	"github.com/cryptopunkscc/astrald/mod/nodes"
 	"github.com/cryptopunkscc/astrald/mod/shell"
-	"slices"
 )
 
 type opStreamsArgs struct {

mod/udp/README.md

@@ -0,0 +1,28 @@
+# UDP Module
+
+This module implements UDP-based communication for the Astrald platform, enabling fast, connectionless data transfer between nodes.
+
+## Current Proof of Concept (PoC) State
+- Basic UDP packet send/receive functionality
+- Simple endpoint resolution
+- Fragmentation and reassembly of large packets
+- Minimal handshake and connection logic
+- Configuration via `config.go`
+
+## Key Components
+- **Dial:** Initiates outbound UDP connections. Integrates with Astral's context (`astral.Context`), endpoint abstraction (`exonet.Endpoint`), and parses endpoints using UDP utilities. Returns a reliable connection for Astral's exonet system. Similar to TCP's Dial, but uses UDP-specific logic and interacts with Astral's node and identity systems.
+- **ResolveEndpoints:** Resolves available UDP endpoints for a node. Uses Astral's identity system (`astral.Identity`) to verify node identity and returns endpoints as `exonet.Endpoint` via Astral's signal utilities (`sig.ArrayToChan`). Connects with Astral's node and endpoint management.
+- **Loader:** Loads the UDP module into Astral. Connects with Astral's node (`astral.Node`), asset management (`core/assets`), and logging (`astral/log.Logger`). Loads configuration from assets, parses public endpoints, and registers the module with Astral's core module system for lifecycle management.
+- **Unpack:** Handles packet reassembly for Astral's exonet system. Uses UDP endpoint parsing and error handling, returning endpoints for Astral's network abstraction. Connects with Astral's error and endpoint utilities.
+- **Server:** Listens for incoming UDP connections. Uses Astral's context and logging, manages connections and server lifecycle, and integrates with Astral's configuration and endpoint parsing. Registers endpoints and connections with Astral's node and router systems.
+- 
+## Possible Improvements for Production Readiness
+- Advanced flow control mechanisms (dynamic window sizing, congestion control)
+- Fast retransmissions and selective acknowledgments (SACK)
+- Adaptive retransmission timers (RTT estimation)
+- Performance optimizations (buffering, batching)
+- Support for NAT traversal and hole punching
+- Comprehensive test coverage
+- Documentation and usage examples
+
+This module is currently a proof of concept and not recommended for production use without further development.

mod/udp/endpoint.go

@@ -0,0 +1,154 @@
+package udp
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"net"
+	"strconv"
+
+	"github.com/cryptopunkscc/astrald/astral"
+	"github.com/cryptopunkscc/astrald/astral/term"
+	"github.com/cryptopunkscc/astrald/mod/exonet"
+)
+
+var _ exonet.Endpoint = &Endpoint{}
+var _ astral.Object = &Endpoint{}
+
+// NOTE: this is same for UDP and TCP - consider moving to a common package
+
+// Endpoint is an astral.Object that holds information about a UDP endpoint,
+// i.e. an IP address and a port.
+// Supports JSON and text.
+type Endpoint struct {
+	IP   IP
+	Port astral.Uint16
+}
+
+func (e *Endpoint) ObjectType() string {
+	return "mod.udp.endpoint"
+}
+
+func (e Endpoint) WriteTo(w io.Writer) (n int64, err error) {
+	return astral.Struct(e).WriteTo(w)
+}
+
+func (e *Endpoint) ReadFrom(r io.Reader) (n int64, err error) {
+	return astral.Struct(e).ReadFrom(r)
+}
+
+// exonet.Endpoint
+
+func (e *Endpoint) Address() string {
+	return net.JoinHostPort(e.IP.String(), strconv.Itoa(int(e.Port)))
+}
+
+func (e *Endpoint) Network() string {
+	return "udp"
+}
+
+// HostString returns the IP address as a string
+func (e *Endpoint) HostString() string {
+	return e.IP.String()
+}
+
+// PortNumber returns the port number as an int
+func (e *Endpoint) PortNumber() int {
+	return int(e.Port)
+}
+
+func (e *Endpoint) Pack() []byte {
+	var b = &bytes.Buffer{}
+	if _, err := e.WriteTo(b); err != nil {
+		return nil
+	}
+	return b.Bytes()
+}
+
+// Text marshaling
+
+func (e Endpoint) MarshalText() (text []byte, err error) {
+	return []byte(e.Address()), nil
+}
+
+func (e *Endpoint) UnmarshalText(text []byte) error {
+	h, p, err := net.SplitHostPort(string(text))
+	if err != nil {
+		return err
+	}
+
+	ip, err := ParseIP(h)
+	if err != nil {
+		return err
+	}
+
+	port, err := strconv.Atoi(p)
+	if err != nil {
+		return err
+	}
+
+	// check if port fits in 16 bits
+	if (port >> 16) > 0 {
+		return errors.New("port out of range")
+	}
+
+	e.IP = ip
+	e.Port = astral.Uint16(port)
+
+	return nil
+}
+
+// ...
+
+func (e *Endpoint) String() string {
+	return e.Address()
+}
+
+func (e *Endpoint) IsZero() bool {
+	return e == nil || e.IP == nil
+}
+
+func ParseEndpoint(s string) (*Endpoint, error) {
+	hostStr, portStr, err := net.SplitHostPort(s)
+	if err != nil {
+		return nil, err
+	}
+
+	ip, err := ParseIP(hostStr)
+	if err != nil {
+		return nil, err
+	}
+
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		return nil, err
+	}
+
+	// check if port fits in 16 bits
+	if (port >> 16) > 0 {
+		return nil, errors.New("port out of range")
+	}
+
+	return &Endpoint{
+		IP:   ip,
+		Port: astral.Uint16(port),
+	}, nil
+}
+
+func (e *Endpoint) UDPAddr() *net.UDPAddr {
+	return &net.UDPAddr{
+		IP:   net.ParseIP(e.IP.String()),
+		Port: int(e.Port),
+	}
+}
+
+func init() {
+	_ = astral.DefaultBlueprints.Add(&Endpoint{})
+
+	term.SetTranslateFunc(func(o *Endpoint) astral.Object {
+		return &term.ColorString{
+			Color: term.HighlightColor,
+			Text:  astral.String32(o.String()),
+		}
+	})
+}

mod/udp/errors.go

@@ -0,0 +1,19 @@
+package udp
+
+import "errors"
+
+var (
+	ErrListenerClosed              = errors.New("listener closed")
+	ErrRetransmissionLimitExceeded = errors.New(
+		"retransmissions limit exceeded")
+
+	// ErrDataLost is emitted on close if there was still buffered or unacked data.
+	ErrDataLost                 = errors.New("unsent data lost on close")
+	ErrPacketTooShort           = errors.New("packet too short")
+	ErrConnClosed               = errors.New("connection closed")
+	ErrInvalidPayloadLength     = errors.New("invalid payload length")
+	ErrZeroMSS                  = errors.New("invalid MSS")
+	ErrMalformedPacket          = errors.New("malformed packet")
+	ErrHandshakeTimeout         = errors.New("handshake timeout")
+	ErrConnectionNotEstablished = errors.New("connection not established")
+)

mod/udp/ip.go

@@ -0,0 +1,97 @@
+package udp
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net"
+
+	"github.com/cryptopunkscc/astrald/astral"
+)
+
+// NOTE: this is same for tcp / udp modules, consider moving to a common package
+type IP net.IP
+
+func ParseIP(s string) (IP, error) {
+	return IP(net.ParseIP(s)), nil
+}
+
+// astral
+
+func (IP) ObjectType() string {
+	return "mod.udp.ip_address"
+}
+
+func (ip IP) WriteTo(w io.Writer) (n int64, err error) {
+	if ip.IsIPv4() {
+		return astral.Bytes8(net.IP(ip).To4()).WriteTo(w)
+	}
+
+	return astral.Bytes8(ip).WriteTo(w)
+}
+
+func (ip *IP) ReadFrom(r io.Reader) (n int64, err error) {
+	return (*astral.Bytes8)(ip).ReadFrom(r)
+}
+
+// json
+
+func (ip IP) MarshalJSON() ([]byte, error) {
+	return json.Marshal(ip.String())
+}
+
+func (ip *IP) UnmarshalJSON(b []byte) error {
+	var str string
+	err := json.Unmarshal(b, &str)
+	if err != nil {
+		return nil
+	}
+
+	parsed := IP(net.ParseIP(str))
+
+	if parsed == nil {
+		return errors.New("invalid IP")
+	}
+
+	*ip = parsed
+	return nil
+}
+
+// text
+
+func (ip IP) MarshalText() (text []byte, err error) {
+	return []byte(ip.String()), nil
+}
+
+func (ip *IP) UnmarshalText(text []byte) error {
+	parsed := IP(net.ParseIP(string(text)))
+	if parsed == nil {
+		return errors.New("invalid IP")
+	}
+	*ip = parsed
+	return nil
+}
+
+// ...
+
+func (ip IP) IsIPv4() bool {
+	return net.IP(ip).To4() != nil
+}
+
+func (ip IP) IsIPv6() bool {
+	return net.IP(ip).To16() != nil
+}
+
+func (ip IP) IsLoopback() bool { return net.IP(ip).IsLoopback() }
+
+func (ip IP) IsGlobalUnicast() bool { return net.IP(ip).IsGlobalUnicast() }
+
+func (ip IP) IsPrivate() bool { return net.IP(ip).IsPrivate() }
+
+func (ip IP) String() string {
+	return net.IP(ip).String()
+}
+
+func init() {
+	_ = astral.DefaultBlueprints.Add(&IP{})
+}

mod/udp/module.go

@@ -0,0 +1,14 @@
+package udp
+
+import (
+	"github.com/cryptopunkscc/astrald/mod/exonet"
+)
+
+const ModuleName = "udp"
+
+type Module interface {
+	exonet.Dialer
+	exonet.Unpacker
+	exonet.Parser
+	ListenPort() int
+}