Gossip Protocol Library for Dart #

A Dart library for implementing gossip-based distributed event synchronization. This library provides the core logic for gossip protocols, allowing nodes to exchange and synchronize events in a distributed system while remaining transport-agnostic.

Features #

🔄 Event-based gossip protocol with vector clock causality tracking
🔌 Transport-agnostic design - bring your own networking layer
💾 Pluggable storage backends via abstract interfaces
⚙️ Configurable gossip behavior and timing parameters
🛡️ Built-in duplicate detection and event ordering
📊 Event statistics and monitoring capabilities
🎯 Peer selection strategies (random, round-robin, least-recent, most-reliable)
🔧 Anti-entropy mechanisms for enhanced consistency

Installation #

Add this to your pubspec.yaml:

dependencies:
  gossip:
    git: https://github.com/da1nerd/gossip.git

Then run:

dart pub get

Quick Start #

Basic Usage #

import 'package:gossip/gossip.dart';

// Create a gossip node with configuration
final config = GossipConfig(
  nodeId: 'node-1',
  gossipInterval: Duration(seconds: 1),
  fanout: 3,
);

final node = GossipNode(
  config: config,
  eventStore: MemoryEventStore(),
  transport: YourTransportImplementation(),
);

// Start the node
await node.start();

// Create and broadcast events
await node.createEvent({
  'type': 'user_action',
  'action': 'login',
  'user': 'alice',
});

// Add peers
node.addPeer(GossipPeer(
  id: 'node-2',
  address: 'http://node2.example.com:8080',
));

// Listen to events
node.onEventReceived.listen((event) {
  print('Received: ${event.payload} from ${event.nodeId}');
});

Transport Implementation #

The library is transport-agnostic. You need to implement the GossipTransport interface:

class HttpTransport implements GossipTransport {
  @override
  Future<void> initialize() async {
    // Set up HTTP server/client
  }

  @override
  Future<GossipDigestResponse> sendDigest(
    GossipPeer peer,
    GossipDigest digest, {
    Duration? timeout,
  }) async {
    // Send HTTP request with digest
    // Return peer's response
  }

  // Implement other methods...
}

Custom Event Store #

Implement your own storage backend:

class DatabaseEventStore implements EventStore {
  @override
  Future<void> saveEvent(Event event) async {
    // Save to your database
  }

  @override
  Future<List<Event>> getEventsSince(
    String nodeId,
    int afterTimestamp, {
    int? limit,
  }) async {
    // Query your database
  }

  // Implement other methods...
}

Core Concepts #

Events #

Events are the fundamental unit of data synchronized across nodes:

final event = Event(
  id: 'unique-event-id',
  nodeId: 'originating-node',
  timestamp: 42,  // Logical timestamp from vector clock
  creationTimestamp: 1640995200000,  // Wall-clock time
  payload: {'user': 'alice', 'action': 'login'},
);

Vector Clocks #

Vector clocks track causality between events across distributed nodes:

final clock = VectorClock();
clock.increment('node-1');  // Increment local timestamp
clock.merge(otherClock);    // Merge knowledge from another node

// Compare clocks for causality
final relationship = clock.compareTo(otherClock);
// Returns: ClockComparison.before, .after, .concurrent, or .equal

Gossip Protocol #

The library implements a 3-phase gossip protocol:

Digest Exchange: Node A sends its vector clock summary to Node B
Event Exchange: Node B responds with missing events and requests
Final Sync: Node A sends the requested events to Node B

Configuration Options #

Basic Configuration #

final config = GossipConfig(
  nodeId: 'my-node',
  gossipInterval: Duration(seconds: 1),    // How often to gossip
  fanout: 3,                               // Number of peers per round
  gossipTimeout: Duration(seconds: 10),    // Timeout for exchanges
  maxEventsPerMessage: 100,                // Batch size limit
  maxMessageSizeBytes: 1024 * 1024,        // 1MB message limit
);

Preset Configurations #

// High-throughput configuration
final config = GossipConfig.highThroughput(
  nodeId: 'fast-node',
  gossipInterval: Duration(milliseconds: 500),
  fanout: 5,
);

// Low-resource configuration
final config = GossipConfig.lowResource(
  nodeId: 'resource-constrained',
  gossipInterval: Duration(seconds: 5),
  fanout: 2,
);

Peer Selection Strategies #

final config = GossipConfig(
  nodeId: 'my-node',
  peerSelectionStrategy: PeerSelectionStrategy.random,
  // Options: random, roundRobin, leastRecentlyContacted, mostReliable
);

Advanced Features #

Event Streams #

Subscribe to various event streams for monitoring and integration:

// Events created by this node
node.onEventCreated.listen((event) {
  print('Created: ${event.payload}');
});

// Events received from other nodes
node.onEventReceived.listen((event) {
  print('Received: ${event.payload} from ${event.nodeId}');
});

// Peer management
node.onPeerAdded.listen((peer) {
  print('Peer added: ${peer.id}');
});

// Gossip exchange results
node.onGossipExchange.listen((result) {
  print('Exchange with ${result.peer.id}: '
        '${result.success ? 'SUCCESS' : 'FAILED'}');
});

Manual Gossip Triggers #

// Trigger gossip with random peers
await node.gossip();

// Gossip with specific peer
final peer = GossipPeer(id: 'node-2', address: 'http://node2:8080');
final result = await node.gossipWith(peer);

print('Exchanged ${result.eventsExchanged} events in ${result.duration}');

Event Store Statistics #

final stats = await eventStore.getStats();
print('Total events: ${stats.totalEvents}');
print('Unique nodes: ${stats.uniqueNodes}');
print('Size: ${stats.sizeInBytes} bytes');

Examples #

See the /example directory for complete working examples:

Simple Example: Basic three-node gossip network with in-memory transport
HTTP Transport: RESTful HTTP-based transport implementation
Database Storage: PostgreSQL-backed event storage
Monitoring: Integration with metrics and logging systems

Architecture #

The library follows a modular architecture:

┌─────────────────┐
│   Application   │
├─────────────────┤
│   GossipNode    │  ← Main coordination logic
├─────────────────┤
│ EventStore │ Transport │  ← Pluggable backends
├─────────────────┤
│ VectorClock │ Event │  ← Core data structures
└─────────────────┘

Best Practices #

Network Transport #

Implement proper timeout handling
Use connection pooling for HTTP transports
Handle network partitions gracefully
Implement exponential backoff for retries

Storage Backend #

Use transactions for consistency
Implement proper indexing for timestamp queries
Consider event retention policies
Monitor storage growth

Configuration #

Tune gossipInterval based on network conditions
Set fanout to 3-5 for optimal convergence
Configure maxEventsPerMessage based on network MTU
Enable anti-entropy for critical consistency requirements

Error Handling #

try {
  await node.createEvent({'data': 'important'});
} on InvalidEventException catch (e) {
  print('Event creation failed: ${e.message}');
} on NodeNotInitializedException catch (e) {
  print('Node not ready: ${e.message}');
}

Testing #

Run tests with:

dart test

The test suite includes:

Unit tests for all core components
Integration tests with mock transport
Property-based tests for vector clock operations
Performance benchmarks

Performance Characteristics #

Convergence Time: O(log N) rounds for N nodes
Message Complexity: O(fanout) messages per round per node
Memory Usage: O(events × nodes) for vector clocks
Network Bandwidth: Configurable via message size limits

Contributing #

Fork the repository at https://github.com/da1nerd/gossip
Create a feature branch (git checkout -b feature/awesome-feature)
Add tests for your changes
Ensure all tests pass (dart test)
Commit your changes (git commit -am 'Add awesome feature')
Push to the branch (git push origin feature/awesome-feature)
Create a Pull Request on GitHub

License #

This project is licensed under the MIT License - see the LICENSE file for details.

gossip 1.0.12 gossip: ^1.0.12 copied to clipboard

Metadata