Chunk group assignment

common/testutils/test_utils.go

@@ -83,3 +83,8 @@ func ExecuteWithTimeout(f func(), duration time.Duration, debugInfo ...any) {
 		panic(fmt.Sprintf(debugInfo[0].(string), debugInfo[1:]...))
 	}
 }
+
+// RandomTime returns a random time.
+func RandomTime() time.Time {
+	return time.Unix(int64(rand.Int31()), int64(rand.Intn(int(time.Second))))
+}

lightnode/chunkgroup/README.md

@@ -0,0 +1,95 @@
+# Chunk Groups
+
+A "chunk group" is a collection of light nodes who are interested in sampling all chunks with a particular chunk index.
+This document describes the algorithm for mapping each light node onto its chunk group.
+
+# Desired Properties
+
+The following properties are desired for the chunk group algorithm:
+
+- **Randomness**: A light node operator should not be able to choose their chunk group prior to registration.
+  A light node should have equal chance of being in any chunk particular chunk group.
+- **Churn**: The chunk group of a light node should change over time.
+- **Stability**: A light node should not change chunk groups too frequently, and at any point in time
+  only a few light nodes should be be changing chunk groups.
+- **Determinism**: The chunk group of a light node should be deterministic based on the node's  
+  seed and the current time. Any two parties should agree which chunk group a light node is in
+  at a particular timestamp.
+
+# Terms
+
+```
+Consider the timeline below, with time moving from left to right.
+
+                              The "+" marks represent                    The 7th time this
+   The genesis time,          the time when a particular                 light node is shuffled.
+   i.e. protocol start        light node is shuffled.                              |
+          |                            |                                           ↓
+          ↓      1          2          ↓          4          5          6          7          8          9
+          |------+---|------+---|------+---|------+---|------+---|------+---|------+---|------+---|------+---|
+          \          /          \      /                                \         /\         /\         /
+           \        /            \    /                                  \       /  \       /  \       /
+            \      /              \  /                                    \     /    \     /    \     /
+             \    /                \/                                      \   /      \   /      \   /
+              \  /                The "shuffle offset".                     \ /        \ /        \ /
+               \/                 Each light node has a                   epoch 6     epoch 7    epoch 8
+  A "shuffle period". Each node   random offset assigned
+  changes chunk groups once per   at registration time.
+  shuffle period. Each shuffle
+  period is marked with a "|".
+```
+
+- **Genesis Time**: The time at which the protocol started. A light node's chunk group is only defined after the genesis
+  time.
+- **Shuffle Period**: The time that should pass in between a particular light node changing chunk groups (e.g. 1 week).
+  Each time one shuffle period passes, all light nodes will be randomly reassigned to a new chunk group.
+- **Shuffle Offset**: In order to avoid too many nodes changing chunk groups at the same time, each node switches chunk
+  groups at an offset relative to the beginning of each shuffle period. This offset is randomly assigned to each node at
+  registration time, but remains constant for the lifetime of the node.
+- **Epoch**: An epoch describes the number of times a particular light node has changed chunk groups. At genesis,
+  all light nodes are in epoch 0. The epoch for a particular light node is incremented by 1 for each time it is
+  randomly shuffled into a new chunk group. The length of each epoch is equal to the shuffle period.
+
+# Algorithm
+
+The algorithm for determining which chunk group a particular light node is described below. A reference implementation
+of this algorithm can be found in [calculations.go](./calculations.go).
+
+## Determining a node's seed
+
+A light node's seed is an 8 byte value that is randomly assigned to the node at registration time. The seed should
+be generated using on-chain randomness that is difficult for an attacker to predict. The light node's seed
+is public information and stored on-chain.
+
+## `randomInt(seed)`
+
+Define a function `randomInt(seed)` that takes an 8 byte unsigned integer as a seed and returns a pseudo-random 
+8 byte unsigned integer. 
+
+- Copy the 8 byte unsigned integer `seed` into an 8 byte array `seedBytes` in big endian order.
+- Use the `seedBytes` as the input to `keccak256` to generate a 32 byte array called `hashBytes`.
+- Use the first 8 bytes of `hashBytes` to create an 8 byte unsigned integer using big endian order called `result`.
+- Return `result`.
+
+## Determining a node's shuffle offset
+
+A node's shuffle offset is a duration between 0 and the shuffle period at nanosecond granularity.
+The shuffle offset is determined as follows:
+
+```
+nodeOffset_nanoseconds := randomInt(nodeSeed) % shufflePeriod_nanoseconds
+```
+
+## Epoch Calculation
+
+At genesis, the epoch for each node is defined as `0`. The epoch increases to `1` at time equal
+to `genesis + nodeOffset`, and then increases by one for each time the clock increases by a further `shufflePeriod`.
+
+## Chunk Group Calculation
+
+To determine a node's chunk group, first compute the current epoch for the node, then plug it into the following
+function:
+
+```
+chunkGroup := randomInt(nodeSeed ^ nodeEpoch) % numberOfChunks
+```

lightnode/chunkgroup/assignment.go

@@ -0,0 +1,46 @@
+package chunkgroup
+
+import (
+	"github.com/Layr-Labs/eigenda/lightnode"
+	"time"
+)
+
+// assignment is a struct that holds a registration and the chunk group it is currently assigned to.
+type assignment struct {
+
+	// registration contains publicly known information about a light node that is registered on-chain.
+	registration *lightnode.Registration
+
+	// shuffleOffset is the offset at which a light node should be shuffled into a new chunk group relative
+	// the beginning of each shuffle interval. This is a function of the light node's seed and the shuffle period
+	// and does not change, so we cache it here.
+	shuffleOffset time.Duration
+
+	// chunkGroup is the chunk group that the light node is currently assigned to.
+	chunkGroup uint32
+
+	// startOfEpoch is the start of the current shuffle epoch,
+	// i.e. the time when this light node was last shuffled into the current chunk group.
+	startOfEpoch time.Time
+
+	// endOfEpoch is the end of the current shuffle epoch,
+	// i.e. the next time when this light node will be shuffled into a new chunk group.
+	endOfEpoch time.Time
+}
+
+// newAssignment creates a new assignment.
+func newAssignment(
+	registration *lightnode.Registration,
+	shuffleOffset time.Duration,
+	chunkGroup uint32,
+	startOfEpoch time.Time,
+	endOfEpoch time.Time) *assignment {
+
+	return &assignment{
+		registration:  registration,
+		shuffleOffset: shuffleOffset,
+		chunkGroup:    chunkGroup,
+		startOfEpoch:  startOfEpoch,
+		endOfEpoch:    endOfEpoch,
+	}
+}

lightnode/chunkgroup/assignment_queue.go

@@ -0,0 +1,158 @@
+package chunkgroup
+
+import (
+	"container/heap"
+	"fmt"
+)
+
+// assignmentHeap implements the heap.Interface for assignment objects, used to create a priority queue.
+type assignmentHeap struct {
+	data []*assignment
+}
+
+// Len returns the number of elements in the priority queue.
+func (h *assignmentHeap) Len() int {
+	return len(h.data)
+}
+
+// Less returns whether the element with index i should sort before the element with index j.
+// This assignmentHeap sorts based on the endOfEpoch of the light nodes.
+func (h *assignmentHeap) Less(i int, j int) bool {
+	ii := h.data[i]
+	jj := h.data[j]
+	return ii.endOfEpoch.Before(jj.endOfEpoch)
+}
+
+// Swap swaps the elements with indexes i and j.
+func (h *assignmentHeap) Swap(i int, j int) {
+	h.data[i], h.data[j] = h.data[j], h.data[i]
+}
+
+// Push adds an element to the end of the priority queue.
+func (h *assignmentHeap) Push(x any) {
+	h.data = append(h.data, x.(*assignment))
+}
+
+// Pop removes and returns the last element in the priority queue.
+func (h *assignmentHeap) Pop() any {
+	n := len(h.data)
+	x := h.data[n-1]
+	h.data = h.data[:n-1]
+	return x
+}
+
+// assignmentQueue is a priority queue that sorts light nodes based on their endOfEpoch.
+type assignmentQueue struct {
+	// The heap that stores the light nodes. Nodes are sorted by their endOfEpoch.
+	heap *assignmentHeap
+
+	// A set of node IDs in the queue. This is used to do efficient removals.
+	// A true value indicates that the node is in the queue. A false value indicates
+	// that the node was removed from the queue but has not yet been fully deleted.
+	nodeIdSet map[uint64]bool
+
+	// The number of elements in the queue. Tracked separately since the heap and NodeIdSet
+	// may contain removed nodes that have not yet been fully garbage collected.
+	size uint
+}
+
+// newAssignmentQueue creates a new priority queue.
+func newAssignmentQueue() *assignmentQueue {
+	return &assignmentQueue{
+		heap: &assignmentHeap{
+			data: make([]*assignment, 0),
+		},
+		nodeIdSet: make(map[uint64]bool),
+	}
+}
+
+// Size returns the number of elements in the priority queue.
+func (queue *assignmentQueue) Size() uint {
+	return queue.size
+}
+
+// Push adds an assignment to the priority queue. This is a no-op if the assignment is already in the queue.
+func (queue *assignmentQueue) Push(assignment *assignment) {
+	notRemoved, present := queue.nodeIdSet[assignment.registration.ID()]
+	if present && notRemoved {
+		return
+	}
+
+	queue.size++
+
+	if !present {
+		heap.Push(queue.heap, assignment)
+	}
+
+	queue.nodeIdSet[assignment.registration.ID()] = true
+}
+
+// Pop removes and returns the assignment with the earliest endOfEpoch.
+func (queue *assignmentQueue) Pop() *assignment {
+	queue.collectGarbage()
+	if queue.size == 0 {
+		return nil
+	}
+	assignment := heap.Pop(queue.heap).(*assignment)
+	delete(queue.nodeIdSet, assignment.registration.ID())
+	queue.size--
+	return assignment
+}
+
+// Peek returns the assignment with the earliest endOfEpoch without removing it from the queue. Returns
+// nil if the queue is empty.
+func (queue *assignmentQueue) Peek() *assignment {
+	queue.collectGarbage()
+	if queue.size == 0 {
+		return nil
+	}
+	return queue.heap.data[0]
+}
+
+// Remove removes the light node with the given ID from the priority queue.
+// This is a no-op if the light node is not in the queue.
+func (queue *assignmentQueue) Remove(lightNodeId uint64) {
+	// Deletion is lazy. The node is fully removed when it reaches the top of the heap.
+
+	notRemoved, present := queue.nodeIdSet[lightNodeId]
+	if !present || !notRemoved {
+		// Element is either not in the queue or has already been marked for removal.
+		return
+	}
+
+	queue.size--
+
+	queue.nodeIdSet[lightNodeId] = false
+}
+
+// collectGarbage removes all nodes that have been removed from the queue but have not yet been fully deleted.
+// This is done by popping elements from the heap until the first element is not marked for deletion.
+func (queue *assignmentQueue) collectGarbage() {
+	if len(queue.heap.data) == 0 {
+		return
+	}
+
+	// sanity check to prevent infinite loops
+	maxIterations := len(queue.heap.data)
+
+	for {
+		maxIterations--
+		if maxIterations < 0 {
+			panic("garbage collection did not terminate")
+		}
+
+		next := queue.heap.data[0]
+
+		notRemoved, present := queue.nodeIdSet[next.registration.ID()]
+		if !present {
+			panic(fmt.Sprintf("node %d is not in the nodeIdSet", next.registration.ID()))
+		}
+
+		if notRemoved {
+			// Once we find the first element that is not marked for deletion, we can stop.
+			return
+		}
+
+		heap.Pop(queue.heap)
+	}
+}