feat(streamer): distribute rewards immediately in the current block

app/keepers/keepers.go

@@ -565,8 +565,8 @@ func (a *AppKeepers) SetupHooks() {
 	a.EpochsKeeper.SetHooks(
 		epochstypes.NewMultiEpochHooks(
 			// insert epochs hooks receivers here
+			a.StreamerKeeper.Hooks(), // x/streamer must be before x/incentives
 			a.IncentivesKeeper.Hooks(),
-			a.StreamerKeeper.Hooks(),
 			a.TxFeesKeeper.Hooks(),
 			a.DelayedAckKeeper.GetEpochHooks(),
 			a.DymNSKeeper.GetEpochHooks(),

proto/dymensionxyz/dymension/streamer/stream.proto

@@ -19,7 +19,7 @@ message Stream {
   uint64 id = 1;
 
   // distribute_to is the distr_info.
-  DistrInfo distribute_to = 2;
+  DistrInfo distribute_to = 2 [ (gogoproto.nullable) = false ];
 
   // coins is the total amount of coins that have been in the stream
   // Can distribute multiple coin denoms

utils/cache/linked_list.go

@@ -0,0 +1,72 @@
+package cache
+
+// List is a double linked list with the limited number of methods.
+// NOT thread safe!
+type List[T any] struct {
+	first *Node[T]
+	last  *Node[T]
+}
+
+func NewList[T any]() *List[T] {
+	return &List[T]{
+		first: nil,
+		last:  nil,
+	}
+}
+
+// Node is an element of doubly linked list.
+// NOT thread safe!
+type Node[T any] struct {
+	elem T
+	prev *Node[T]
+	next *Node[T]
+	list *List[T]
+}
+
+// Insert inserts new node to the end of the list.
+// NOT thread safe!
+func (l *List[T]) Insert(elem T) *Node[T] {
+	n := &Node[T]{
+		elem: elem,
+		prev: l.last,
+		next: nil,
+		list: l,
+	}
+	if l.last == nil {
+		l.first = n
+	} else {
+		l.last.next = n
+	}
+	l.last = n
+	return n
+}
+
+// Delete removes node from the list.
+// NOT thread safe!
+func (n *Node[T]) Delete() {
+	if n.list.first == n {
+		n.list.first = n.next
+	}
+	if n.list.last == n {
+		n.list.last = n.prev
+	}
+	if n.next != nil {
+		n.next.prev = n.prev
+	}
+	if n.prev != nil {
+		n.prev.next = n.next
+	}
+	n.next, n.prev = nil, nil
+}
+
+// Range loops over all elements and calls f with each of them.
+// NOT thread safe!
+func (l *List[T]) Range(f func(T) bool) {
+	n := l.first
+	for n != nil {
+		if !f(n.elem) {
+			break
+		}
+		n = n.next
+	}
+}

utils/cache/linked_list_internal_test.go

@@ -0,0 +1,173 @@
+package cache
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestList(t *testing.T) {
+	t.Parallel()
+
+	t.Run("empty", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+		assertForEach(t, l, nil)
+	})
+	t.Run("insert 1 element", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+
+		a := l.Insert(10)
+
+		assertForEach(t, l, []int{10})
+		assert.Same(t, a, l.first)
+		assert.Same(t, a, l.last)
+		assert.Nil(t, a.prev)
+		assert.Nil(t, a.next)
+	})
+	t.Run("delete 1 element", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+		a := l.Insert(10)
+
+		a.Delete()
+
+		assertForEach(t, l, nil)
+		assert.Nil(t, l.first)
+		assert.Nil(t, l.last)
+		assert.Nil(t, a.prev)
+		assert.Nil(t, a.next)
+	})
+	t.Run("insert 2 elements", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+
+		a := l.Insert(10)
+		b := l.Insert(20)
+
+		assertForEach(t, l, []int{10, 20})
+		assert.Same(t, a, l.first)
+		assert.Same(t, b, l.last)
+		assert.Nil(t, a.prev)
+		assert.Same(t, b, a.next)
+		assert.Same(t, a, b.prev)
+		assert.Nil(t, b.next)
+	})
+	t.Run("delete last from 2 elements", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+		a := l.Insert(10)
+		b := l.Insert(20)
+
+		b.Delete()
+
+		assertForEach(t, l, []int{10})
+		assert.Same(t, a, l.first)
+		assert.Same(t, a, l.last)
+		assert.Nil(t, a.prev)
+		assert.Nil(t, a.next)
+		assert.Nil(t, b.prev)
+		assert.Nil(t, b.next)
+	})
+	t.Run("delete first from 2 elements", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+		a := l.Insert(10)
+		b := l.Insert(20)
+
+		a.Delete()
+
+		assertForEach(t, l, []int{20})
+		assert.Same(t, b, l.first)
+		assert.Same(t, b, l.last)
+		assert.Nil(t, a.prev)
+		assert.Nil(t, a.next)
+		assert.Nil(t, b.prev)
+		assert.Nil(t, b.next)
+	})
+	t.Run("insert 3 elements", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+
+		a := l.Insert(10)
+		b := l.Insert(20)
+		c := l.Insert(30)
+
+		assertForEach(t, l, []int{10, 20, 30})
+		assert.Same(t, a, l.first)
+		assert.Same(t, c, l.last)
+		assert.Nil(t, a.prev)
+		assert.Same(t, b, a.next)
+		assert.Same(t, a, b.prev)
+		assert.Same(t, c, b.next)
+		assert.Same(t, b, c.prev)
+		assert.Nil(t, c.next)
+	})
+	t.Run("delete last from 3 elements", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+		a := l.Insert(10)
+		b := l.Insert(20)
+		c := l.Insert(30)
+
+		c.Delete()
+
+		assertForEach(t, l, []int{10, 20})
+		assert.Same(t, a, l.first)
+		assert.Same(t, b, l.last)
+		assert.Nil(t, a.prev)
+		assert.Same(t, b, a.next)
+		assert.Same(t, a, b.prev)
+		assert.Nil(t, b.next)
+		assert.Nil(t, c.prev)
+		assert.Nil(t, c.next)
+	})
+	t.Run("delete first from 3 elements", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+		a := l.Insert(10)
+		b := l.Insert(20)
+		c := l.Insert(30)
+
+		a.Delete()
+
+		assertForEach(t, l, []int{20, 30})
+		assert.Same(t, b, l.first)
+		assert.Same(t, c, l.last)
+		assert.Nil(t, a.prev)
+		assert.Nil(t, a.next)
+		assert.Nil(t, b.prev)
+		assert.Same(t, c, b.next)
+		assert.Same(t, b, c.prev)
+		assert.Nil(t, c.next)
+	})
+	t.Run("delete middle from 3 elements", func(t *testing.T) {
+		t.Parallel()
+		l := NewList[int]()
+		a := l.Insert(10)
+		b := l.Insert(20)
+		c := l.Insert(30)
+
+		b.Delete()
+
+		assertForEach(t, l, []int{10, 30})
+		assert.Same(t, a, l.first)
+		assert.Same(t, c, l.last)
+		assert.Nil(t, a.prev)
+		assert.Same(t, c, a.next)
+		assert.Nil(t, b.prev)
+		assert.Nil(t, b.next)
+		assert.Same(t, a, c.prev)
+		assert.Nil(t, c.next)
+	})
+}
+
+func assertForEach[T any](t *testing.T, l *List[T], expected []T) {
+	var actual []T
+	l.Range(func(v T) bool {
+		actual = append(actual, v)
+		return true
+	})
+	assert.Equal(t, expected, actual)
+}

utils/cache/ordered.go

@@ -0,0 +1,99 @@
+package cache
+
+// InsertionOrdered cache preserves insertion-order.
+// NOT thread safe!
+type InsertionOrdered[K comparable, V any] struct {
+	dataIdx map[K]*Node[V]
+	key     func(V) K
+	data    *List[V]
+}
+
+func NewInsertionOrdered[K comparable, V any](key func(V) K, initial ...V) *InsertionOrdered[K, V] {
+	cache := &InsertionOrdered[K, V]{
+		dataIdx: make(map[K]*Node[V], 0),
+		key:     key,
+		data:    NewList[V](),
+	}
+	cache.UpsertMultiple(initial...)
+	return cache
+}
+
+func (c *InsertionOrdered[K, V]) Reset(values ...V) {
+	c.dataIdx = make(map[K]*Node[V], len(values))
+	c.data = NewList[V]()
+	c.UpsertMultiple(values...)
+}
+
+func (c *InsertionOrdered[K, V]) Upsert(value V) {
+	key := c.key(value)
+	v, ok := c.dataIdx[key]
+	if ok {
+		// if the value is present, just update it
+		v.elem = value
+	} else {
+		// otherwise, create a new one
+		c.dataIdx[key] = c.data.Insert(value)
+	}
+}
+
+func (c *InsertionOrdered[K, V]) UpsertMultiple(values ...V) {
+	for _, value := range values {
+		c.Upsert(value)
+	}
+}
+
+func (c *InsertionOrdered[K, V]) Delete(keys ...K) {
+	for _, key := range keys {
+		if value, ok := c.dataIdx[key]; ok {
+			value.Delete()
+		}
+		delete(c.dataIdx, key)
+	}
+}
+
+func (c *InsertionOrdered[K, V]) Get(key K) (V, bool) {
+	value, ok := c.dataIdx[key]
+	if ok {
+		return value.elem, true
+	}
+	var zero V
+	return zero, false
+}
+
+func (c *InsertionOrdered[K, V]) GetAll() []V {
+	res := make([]V, 0, len(c.dataIdx))
+	c.data.Range(func(v V) bool {
+		res = append(res, v)
+		return true
+	})
+	return res
+}
+
+func (c *InsertionOrdered[K, V]) Filter(condition func(V) bool) []V {
+	var res []V
+	c.data.Range(func(v V) bool {
+		if condition(v) {
+			res = append(res, v)
+		}
+		return true
+	})
+	return res
+}
+
+func (c *InsertionOrdered[K, V]) FindFirst(condition func(V) bool) (V, bool) {
+	var res V
+	var found bool
+	c.data.Range(func(v V) bool {
+		if condition(v) {
+			res = v
+			found = true
+			return false
+		}
+		return true
+	})
+	return res, found
+}
+
+func (c *InsertionOrdered[K, V]) Range(f func(V) bool) {
+	c.data.Range(f)
+}