Skip to content

Commit 9e90c33

Browse files
mikeramikera
committed
Lattice Queue WIP
1 parent 4642caa commit 9e90c33

File tree

7 files changed

+769
-2
lines changed

7 files changed

+769
-2
lines changed

‎convex-core/src/main/java/convex/core/cvm/Keywords.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -156,8 +156,9 @@ public class Keywords {
156156
public static final Keyword DATA = Keyword.intern("data");
157157
public static final Keyword FS = Keyword.intern("fs");
158158
public static final Keyword KV = Keyword.intern("kv");
159-
160-
159+
public static final Keyword QUEUE = Keyword.intern("queue");
160+
161+
161162
// General API keywords
162163
public static final Keyword FAUCET = Keyword.intern("faucet");
163164

‎convex-core/src/main/java/convex/lattice/Lattice.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import convex.lattice.generic.MapLattice;
1010
import convex.lattice.generic.OwnerLattice;
1111
import convex.lattice.kv.KVStoreLattice;
12+
import convex.lattice.queue.TopicLattice;
1213

1314
/**
1415
* Static utility base for the lattice
@@ -22,6 +23,8 @@ public class Lattice {
2223
* where drive names are AString (not Keywords)
2324
* - :kv - Key-value databases (db name -> owner/node -> signed KV store)
2425
* Each database has per-node signed replicas merged via KVStoreLattice
26+
* - :queue - Kafka-style message queues (owner -> topic -> :partitions -> partition id -> queue)
27+
* Each topic has metadata + partitions; each partition is an append-only log
2528
*/
2629
public static KeyedLattice ROOT = KeyedLattice.create(
2730
Keywords.DATA, DataLattice.INSTANCE,
@@ -30,6 +33,9 @@ public class Lattice {
3033
),
3134
Keywords.KV, OwnerLattice.create(
3235
MapLattice.create(KVStoreLattice.INSTANCE)
36+
),
37+
Keywords.QUEUE, OwnerLattice.create(
38+
MapLattice.create(TopicLattice.INSTANCE)
3339
)
3440
);
3541
}

‎convex-core/src/main/java/convex/lattice/queue/LatticeMQ.java‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
package convex.lattice.queue;
2+
3+
import convex.core.data.ACell;
4+
import convex.core.data.Index;
5+
import convex.core.data.Keyword;
6+
import convex.core.data.Strings;
7+
import convex.lattice.cursor.ALatticeCursor;
8+
import convex.lattice.cursor.Cursors;
9+
import convex.lattice.generic.MapLattice;
10+
11+
/**
12+
* A Kafka-style distributed message queue system built on Convex lattice technology.
13+
*
14+
* <p>Provides a two-level hierarchy of <b>topics</b> and <b>partitions</b>. Each topic
15+
* contains one or more partitions, and each partition is an independent append-only log
16+
* ({@link LatticeQueue}).</p>
17+
*
18+
* <p>In the full lattice hierarchy, the path from root is:
19+
* {@code [:queue, <owner>, :value, <topic-name>, <partition-id>]}</p>
20+
*
21+
* <h2>Lattice Hierarchy</h2>
22+
* <pre>
23+
* LatticeMQ (topics map)
24+
* └─ LatticeTopic (partitions map)
25+
* └─ LatticeQueue (single partition — append-only log)
26+
* </pre>
27+
*
28+
* <h2>Usage</h2>
29+
* <pre>{@code
30+
* LatticeMQ mq = LatticeMQ.create();
31+
*
32+
* // Access a topic and partition
33+
* LatticeQueue q = mq.topic("user-events").partition(0);
34+
* q.offer(Strings.create("event data"));
35+
*
36+
* // Or directly
37+
* mq.partition("user-events", 0).offer(Strings.create("event data"));
38+
* }</pre>
39+
*/
40+
public class LatticeMQ {
41+
42+
/**
43+
* The lattice definition for the topic map level:
44+
* MapLattice(topic-name → TopicLattice(partitions + metadata))
45+
*/
46+
static final MapLattice<ACell, Index<Keyword, ACell>> TOPIC_MAP =
47+
MapLattice.create(TopicLattice.INSTANCE);
48+
49+
private final ALatticeCursor<?> cursor;
50+
51+
/**
52+
* Wraps an existing cursor at the topic-map level.
53+
*/
54+
public LatticeMQ(ALatticeCursor<?> cursor) {
55+
this.cursor = cursor;
56+
}
57+
58+
/**
59+
* Creates a new standalone message queue system.
60+
*
61+
* <p>For use within the full lattice hierarchy, obtain a cursor descended to
62+
* {@code [:queue, <owner>, :value]} and pass it to the constructor instead.</p>
63+
*/
64+
public static LatticeMQ create() {
65+
return new LatticeMQ(Cursors.createLattice(TOPIC_MAP));
66+
}
67+
68+
/**
69+
* Returns a topic by name. The topic is created implicitly on first write.
70+
*
71+
* @param name Topic name
72+
* @return Topic handle
73+
*/
74+
public LatticeTopic topic(String name) {
75+
return topic(Strings.create(name));
76+
}
77+
78+
/**
79+
* Returns a topic by CVM key. The topic is created implicitly on first write.
80+
*
81+
* @param name Topic key (typically AString)
82+
* @return Topic handle
83+
*/
84+
public LatticeTopic topic(ACell name) {
85+
return new LatticeTopic(cursor.descend(name));
86+
}
87+
88+
/**
89+
* Convenience method to access a specific partition directly.
90+
*
91+
* @param topic Topic name
92+
* @param partition Partition number
93+
* @return Queue for the specified partition
94+
*/
95+
public LatticeQueue partition(String topic, long partition) {
96+
return topic(topic).partition(partition);
97+
}
98+
99+
/**
100+
* Creates a forked copy of this MQ system for independent operation.
101+
* All topics and partitions are forked together.
102+
*/
103+
@SuppressWarnings("unchecked")
104+
public LatticeMQ fork() {
105+
return new LatticeMQ(((ALatticeCursor<ACell>) cursor).fork());
106+
}
107+
108+
/**
109+
* Syncs this forked MQ system back to its parent, merging all changes
110+
* across all topics and partitions.
111+
*/
112+
@SuppressWarnings("unchecked")
113+
public void sync() {
114+
((ALatticeCursor<ACell>) cursor).sync();
115+
}
116+
117+
/**
118+
* Returns the underlying lattice cursor.
119+
*/
120+
public ALatticeCursor<?> cursor() {
121+
return cursor;
122+
}
123+
}

‎convex-core/src/main/java/convex/lattice/queue/LatticeTopic.java‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,186 @@
1+
package convex.lattice.queue;
2+
3+
import convex.core.data.ACell;
4+
import convex.core.data.AHashMap;
5+
import convex.core.data.AVector;
6+
import convex.core.data.Index;
7+
import convex.core.data.Keyword;
8+
import convex.core.data.prim.CVMLong;
9+
import convex.lattice.cursor.ALatticeCursor;
10+
11+
/**
12+
* Represents a single topic in the lattice message queue system.
13+
*
14+
* <p>A topic contains one or more <b>partitions</b>, each an independent
15+
* {@link LatticeQueue} (append-only log), plus topic-level <b>metadata</b>
16+
* (e.g. partition count, retention policy, ownership).</p>
17+
*
18+
* <p>The topic state is a {@link TopicLattice} value: an {@code Index<Keyword, ACell>}
19+
* with {@code :partitions} (partition map) and {@code :meta} (metadata map).</p>
20+
*
21+
* <h2>Partitioning</h2>
22+
* <p>Producers can write to a specific partition or use {@link #offer(ACell, ACell)}
23+
* to auto-partition by key hash. The number of partitions is stored in topic
24+
* metadata under {@code :num-partitions} and must be set before auto-partitioning.</p>
25+
*
26+
* <h2>Usage</h2>
27+
* <pre>{@code
28+
* LatticeTopic topic = mq.topic("user-events");
29+
*
30+
* // Configure partitions
31+
* topic.setNumPartitions(4);
32+
*
33+
* // Write to a specific partition
34+
* topic.partition(0).offer(Strings.create("event-1"));
35+
*
36+
* // Auto-partition by key hash
37+
* topic.offer(Strings.create("user-42"), Strings.create("login"));
38+
* }</pre>
39+
*/
40+
public class LatticeTopic {
41+
42+
public static final Keyword KEY_NUM_PARTITIONS = Keyword.intern("num-partitions");
43+
44+
private final ALatticeCursor<?> cursor;
45+
46+
/**
47+
* Wraps an existing cursor at the TopicLattice level.
48+
*/
49+
LatticeTopic(ALatticeCursor<?> cursor) {
50+
this.cursor = cursor;
51+
}
52+
53+
/**
54+
* Returns a partition by integer ID. The partition is created implicitly on first write.
55+
*
56+
* @param id Partition number (0-based, like Kafka)
57+
* @return Queue for the specified partition
58+
*/
59+
public LatticeQueue partition(long id) {
60+
return partition(CVMLong.create(id));
61+
}
62+
63+
/**
64+
* Returns a partition by CVM key.
65+
*
66+
* <p>Ensures the topic state is initialised as an {@code Index} before
67+
* descending, so that {@code RT.assocIn} preserves the correct type on
68+
* writeback through the cursor hierarchy.</p>
69+
*
70+
* @param id Partition key (typically CVMLong)
71+
* @return Queue for the specified partition
72+
*/
73+
@SuppressWarnings("unchecked")
74+
public LatticeQueue partition(ACell id) {
75+
// Ensure topic state is Index before descended cursors write through it.
76+
// RT.assocIn creates MapLeaf for null intermediaries; by initialising to
77+
// Index.EMPTY first, subsequent assoc() calls preserve the Index type.
78+
ensureInitialised();
79+
80+
ALatticeCursor<AVector<ACell>> partCursor = cursor.descend(TopicLattice.KEY_PARTITIONS, id);
81+
return new LatticeQueue(partCursor);
82+
}
83+
84+
// ===== Metadata =====
85+
86+
/**
87+
* Gets a topic metadata value by key.
88+
*/
89+
@SuppressWarnings("unchecked")
90+
public ACell getMeta(ACell key) {
91+
ALatticeCursor<Index<Keyword, ACell>> c = (ALatticeCursor<Index<Keyword, ACell>>) cursor;
92+
Index<Keyword, ACell> state = c.get();
93+
AHashMap<ACell, ACell> meta = TopicLattice.getMeta(state);
94+
return meta.get(key);
95+
}
96+
97+
/**
98+
* Sets a topic metadata value.
99+
*/
100+
@SuppressWarnings("unchecked")
101+
public void setMeta(ACell key, ACell value) {
102+
ALatticeCursor<Index<Keyword, ACell>> c = (ALatticeCursor<Index<Keyword, ACell>>) cursor;
103+
c.updateAndGet(state -> {
104+
if (state == null) state = TopicLattice.INSTANCE.zero();
105+
AHashMap<ACell, ACell> meta = TopicLattice.getMeta(state);
106+
return state.assoc(TopicLattice.KEY_META, meta.assoc(key, value));
107+
});
108+
}
109+
110+
/**
111+
* Returns the configured number of partitions, or 0 if not set.
112+
*/
113+
public long getNumPartitions() {
114+
ACell val = getMeta(KEY_NUM_PARTITIONS);
115+
if (val instanceof CVMLong l) return l.longValue();
116+
return 0;
117+
}
118+
119+
/**
120+
* Configures the number of partitions for this topic.
121+
*/
122+
public void setNumPartitions(long n) {
123+
setMeta(KEY_NUM_PARTITIONS, CVMLong.create(n));
124+
}
125+
126+
/**
127+
* Produces a keyed record to an auto-selected partition.
128+
*
129+
* <p>The partition is chosen by hashing the key modulo the configured
130+
* partition count. Records with the same key always go to the same
131+
* partition, preserving per-key ordering (the same guarantee Kafka provides).</p>
132+
*
133+
* <p>Requires {@link #setNumPartitions(long)} to have been called first.</p>
134+
*
135+
* @param key Record key (used for partition selection and stored in the entry)
136+
* @param value Record value
137+
* @return Absolute offset assigned within the selected partition
138+
* @throws IllegalStateException if partition count is not configured
139+
*/
140+
public long offer(ACell key, ACell value) {
141+
long numPartitions = getNumPartitions();
142+
if (numPartitions <= 0) {
143+
throw new IllegalStateException(
144+
"Topic has no partitions configured. Call setNumPartitions() first.");
145+
}
146+
long partId = Math.abs(key.getHash().longValue()) % numPartitions;
147+
return partition(partId).offer(key, value);
148+
}
149+
150+
/**
151+
* Creates a forked copy of this topic for independent operation.
152+
* All partitions within this topic are forked together.
153+
*/
154+
@SuppressWarnings("unchecked")
155+
public LatticeTopic fork() {
156+
return new LatticeTopic(((ALatticeCursor<ACell>) cursor).fork());
157+
}
158+
159+
/**
160+
* Syncs this forked topic back to its parent, merging all partition changes.
161+
*/
162+
@SuppressWarnings("unchecked")
163+
public void sync() {
164+
((ALatticeCursor<ACell>) cursor).sync();
165+
}
166+
167+
/**
168+
* Returns the underlying lattice cursor.
169+
*/
170+
public ALatticeCursor<?> cursor() {
171+
return cursor;
172+
}
173+
174+
/**
175+
* Ensures the topic cursor is initialised to an Index so that
176+
* descended cursors (which write back via RT.assocIn) preserve the type.
177+
*/
178+
@SuppressWarnings("unchecked")
179+
private void ensureInitialised() {
180+
ALatticeCursor<Index<Keyword, ACell>> c = (ALatticeCursor<Index<Keyword, ACell>>) cursor;
181+
c.updateAndGet(state -> {
182+
if (state == null) return TopicLattice.INSTANCE.zero();
183+
return state;
184+
});
185+
}
186+
}

0 commit comments

Comments
 (0)