Skip to content

Commit c276c1a

Browse files
MasterPtatoMasterPtato
committed
fix: get pb manager test working e2e again
1 parent 1ccdcba commit c276c1a

File tree

34 files changed

+14947
-303
lines changed

34 files changed

+14947
-303
lines changed

‎Cargo.toml‎

Lines changed: 4 additions & 1 deletion
Large diffs are not rendered by default.

‎examples/system-test-actor/package.json‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,12 +6,15 @@
66
"dependencies": {
77
"@hono/node-server": "^1.13.8",
88
"@hono/node-ws": "^1.1.0",
9+
"@rivet-gg/runner-protocol": "file:../../sdks/runner-protocol",
10+
"google-protobuf": "^3.21.2",
911
"hono": "^4.6.17"
1012
},
1113
"devDependencies": {
1214
"@rivet-gg/actor-core": "^5.1.2",
1315
"@rivet-gg/api": "^24.6.2",
1416
"@types/deno": "^2.2.0",
17+
"@types/google-protobuf": "^3.15.0",
1518
"@types/node": "^22.13.9",
1619
"@types/ws": "^8.18.0",
1720
"dgram": "^1.0.1",
@@ -21,7 +24,7 @@
2124
"ws": "^8.18.1"
2225
},
2326
"scripts": {
24-
"start": "tsx ws.ts",
27+
"start": "tsx src/main.ts",
2528
"test": "tsx tests/client.ts",
2629
"build": "tsc --outDir dist"
2730
}

‎examples/system-test-actor/src/managerClient.ts‎

Lines changed: 46 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import * as net from "net";
22
import * as fs from "fs";
33
import { setInterval, clearInterval } from "timers";
44
import * as util from "util";
5+
import { encodeFrame, decodeFrames, types as protocol } from "@rivet-gg/runner-protocol";
56

67
export function connectToManager() {
78
const socketPath = process.env.RIVET_MANAGER_SOCKET_PATH;
@@ -25,77 +26,79 @@ export function connectToManager() {
2526

2627
// Start ping loop to keep connection alive
2728
pingInterval = setInterval(() => {
28-
const pingMessage = { ping: null };
29+
const pingMessage = new protocol.ToManager({
30+
ping: new protocol.ToManager.Ping()
31+
});
2932
client.write(encodeFrame(pingMessage));
3033
}, 2000);
3134
});
3235

3336
client.on("data", (data) => {
34-
const packets = decodeFrames(data);
37+
const packets = decodeFrames(data, protocol.ToRunner);
3538

3639
for (let packet of packets) {
3740
console.log("Received packet from manager:", util.inspect(packet, { depth: null }));
3841

3942
if (packet.start_actor) {
40-
const response = {
41-
actor_state_update: {
43+
const response = new protocol.ToManager({
44+
actor_state_update: new protocol.ToManager.ActorStateUpdate({
4245
actor_id: packet.start_actor.actor_id,
4346
generation: packet.start_actor.generation,
44-
state: {
45-
running: null,
46-
},
47-
},
48-
};
47+
state: new protocol.ActorState({
48+
running: new protocol.ActorState.Running()
49+
})
50+
})
51+
});
4952
client.write(encodeFrame(response));
5053

5154
console.log(`actor_${packet.start_actor.actor_id}`, 'fweh');
5255

53-
const kvMessage = {
54-
kv: {
56+
const kvMessage = new protocol.ToManager({
57+
kv: new rivet.pegboard.kv.Request({
5558
actor_id: packet.start_actor.actor_id,
5659
generation: packet.start_actor.generation,
5760
request_id: 1,
58-
data: {
59-
put: {
60-
keys: [
61-
[[1, 2, 3], [4, 5, 6]],
62-
],
63-
values: [
64-
[11, 12, 13, 14, 15, 16]
65-
],
66-
}
67-
}
68-
}
69-
};
61+
put: new rivet.pegboard.kv.Request.Put({
62+
keys: [
63+
new rivet.pegboard.kv.Key({
64+
segments: [new Uint8Array([1, 2, 3]), new Uint8Array([4, 5, 6])]
65+
})
66+
],
67+
values: [
68+
new Uint8Array([11, 12, 13, 14, 15, 16])
69+
]
70+
})
71+
})
72+
});
7073
client.write(encodeFrame(kvMessage));
7174

72-
const kvMessage2 = {
73-
kv: {
75+
const kvMessage2 = new protocol.ToManager({
76+
kv: new rivet.pegboard.kv.Request({
7477
actor_id: packet.start_actor.actor_id,
7578
generation: packet.start_actor.generation,
7679
request_id: 2,
77-
data: {
78-
get: {
79-
keys: [
80-
[[1, 2, 3], [4, 5, 6]]
81-
],
82-
}
83-
}
84-
}
85-
};
80+
get: new rivet.pegboard.kv.Request.Get({
81+
keys: [
82+
new rivet.pegboard.kv.Key({
83+
segments: [new Uint8Array([1, 2, 3]), new Uint8Array([4, 5, 6])]
84+
})
85+
]
86+
})
87+
})
88+
});
8689
client.write(encodeFrame(kvMessage2));
8790
} else if (packet.signal_actor) {
88-
const response = {
89-
actor_state_update: {
91+
const response = new protocol.ToManager({
92+
actor_state_update: new protocol.ToManager.ActorStateUpdate({
9093
actor_id: packet.signal_actor.actor_id,
9194
generation: packet.signal_actor.generation,
92-
state: {
93-
exited: {
94-
exit_code: 0,
95-
},
96-
},
97-
},
98-
};
95+
state: new protocol.ActorState({
96+
exited: new protocol.ActorState.Exited({
97+
exit_code: 0
98+
})
99+
})
100+
})
101+
});
99102
client.write(encodeFrame(response));
100103
}
101104
}
@@ -115,33 +118,3 @@ export function connectToManager() {
115118
});
116119
}
117120

118-
function encodeFrame(payload: any): Buffer {
119-
const json = JSON.stringify(payload);
120-
const payloadLength = Buffer.alloc(4);
121-
payloadLength.writeUInt32BE(json.length, 0);
122-
123-
const header = Buffer.alloc(4); // All zeros for now
124-
125-
return Buffer.concat([payloadLength, header, Buffer.from(json)]);
126-
}
127-
128-
function decodeFrames(buffer: Buffer): any[] {
129-
const packets = [];
130-
let offset = 0;
131-
132-
while (offset < buffer.length) {
133-
if (buffer.length - offset < 8) break; // Incomplete frame length + header
134-
const payloadLength = buffer.readUInt32BE(offset);
135-
offset += 4;
136-
137-
// Skip the header (4 bytes)
138-
offset += 4;
139-
140-
if (buffer.length - offset < payloadLength) break; // Incomplete frame data
141-
const json = buffer.subarray(offset, offset + payloadLength).toString();
142-
packets.push(JSON.parse(json));
143-
offset += payloadLength;
144-
}
145-
146-
return packets;
147-
}

‎examples/system-test-actor/yarn.lock‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -277,13 +277,29 @@ __metadata:
277277
languageName: node
278278
linkType: hard
279279

280+
"@rivet-gg/runner-protocol@file:../../sdks/runner-protocol::locator=system-test-container%40workspace%3A.":
281+
version: 1.0.0
282+
resolution: "@rivet-gg/runner-protocol@file:../../sdks/runner-protocol#../../sdks/runner-protocol::hash=3263e1&locator=system-test-container%40workspace%3A."
283+
dependencies:
284+
google-protobuf: "npm:^3.21.0"
285+
checksum: 10c0/b6c92f6d634eac885af3812bee67f905bbcab9edd7c32558540e0037703bb3dd273754e7282044a47aeedf980dfefb50f7828a144f4c5fe4a6b54099c973d7d8
286+
languageName: node
287+
linkType: hard
288+
280289
"@types/deno@npm:^2.2.0":
281290
version: 2.2.0
282291
resolution: "@types/deno@npm:2.2.0"
283292
checksum: 10c0/cb45bbffe66a3008224a509c6bcb338921cc68b9045363f77ba5d84650d879b8fd4c810db24369a93fbce4a8e2855808bb141c0447feb47d911a7512ba374bde
284293
languageName: node
285294
linkType: hard
286295

296+
"@types/google-protobuf@npm:^3.15.0":
297+
version: 3.15.12
298+
resolution: "@types/google-protobuf@npm:3.15.12"
299+
checksum: 10c0/721783234e627f367dd710c345a1eaa9dca4ac64910032cef0c851c1821e05d06ffb51e9d1693080f1c0797a8674f89130fa56a390395ec7791ea8506d2f3bfb
300+
languageName: node
301+
linkType: hard
302+
287303
"@types/node@npm:*, @types/node@npm:^22.13.9":
288304
version: 22.13.9
289305
resolution: "@types/node@npm:22.13.9"
@@ -847,6 +863,13 @@ __metadata:
847863
languageName: node
848864
linkType: hard
849865

866+
"google-protobuf@npm:^3.21.0, google-protobuf@npm:^3.21.2":
867+
version: 3.21.4
868+
resolution: "google-protobuf@npm:3.21.4"
869+
checksum: 10c0/28f2800f7fe1a8fc55eb58ba76e158268407bfb3b90646eaf8a177dd92a2e522459b773f8132ae546e60ac3b6f5947557a1cf3d963a05bb594f43bcde640f54f
870+
languageName: node
871+
linkType: hard
872+
850873
"gopd@npm:^1.2.0":
851874
version: 1.2.0
852875
resolution: "gopd@npm:1.2.0"
@@ -1533,10 +1556,13 @@ __metadata:
15331556
"@hono/node-ws": "npm:^1.1.0"
15341557
"@rivet-gg/actor-core": "npm:^5.1.2"
15351558
"@rivet-gg/api": "npm:^24.6.2"
1559+
"@rivet-gg/runner-protocol": "file:../../sdks/runner-protocol"
15361560
"@types/deno": "npm:^2.2.0"
1561+
"@types/google-protobuf": "npm:^3.15.0"
15371562
"@types/node": "npm:^22.13.9"
15381563
"@types/ws": "npm:^8.18.0"
15391564
dgram: "npm:^1.0.1"
1565+
google-protobuf: "npm:^3.21.2"
15401566
hono: "npm:^4.6.17"
15411567
node-fetch: "npm:^3.3.2"
15421568
tsx: "npm:^4.7.0"

‎package.json‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
"devDependencies": {
2323
"@biomejs/biome": "^1.9.4",
2424
"@yarnpkg/plugin-exec": "^3.0.1",
25+
"protoc-gen-js": "^3.21.2",
26+
"protoc-gen-ts": "^0.8.7",
2527
"turbo": "^2.0.1"
2628
},
2729
"resolutions": {

‎packages/core/services/cluster/src/workflows/server/install/install_scripts/components/vector.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ pub fn configure(namespace: &str, config: &Config, pool_type: PoolType) -> Globa
7474
// Buffer to disk for durability & reduce memory usage
7575
"buffer": {
7676
"type": "disk",
77-
"max_size": 2 * 1024 * 1024 * 1024, // 2 GiB
77+
"max_size": 2u64 * 1024 * 1024 * 1024, // 2 GiB
7878
"when_full": "block"
7979
}
8080
}

‎packages/edge/infra/client/actor-kv/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ fdb-util.workspace = true
1111
foundationdb.workspace = true
1212
futures-util = { version = "0.3" }
1313
indexmap = { version = "2.0" }
14-
pegboard-config.workspace = true
14+
pegboard-runner-protocol.workspace = true
1515
prost = "0.14"
1616
rivet-util-id.workspace = true
1717
serde = { version = "1.0.195", features = ["derive"] }

‎packages/edge/infra/client/actor-kv/src/entry.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use anyhow::*;
22
use foundationdb as fdb;
3-
use pegboard_config::runner_protocol::proto::kv;
3+
use pegboard_runner_protocol::proto::kv;
44
use prost::Message;
55

66
use crate::key::Key;

‎packages/edge/infra/client/actor-kv/src/key.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use foundationdb::tuple::{
22
Bytes, PackResult, TupleDepth, TuplePack, TupleUnpack, VersionstampOffset,
33
};
4-
use pegboard_config::runner_protocol::proto::kv;
4+
use pegboard_runner_protocol::proto::kv;
55
use prost::Message;
66

77
// TODO: Custom deser impl that uses arrays instead of objects?

‎packages/edge/infra/client/actor-kv/src/lib.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ use indexmap::IndexMap;
1414
pub use key::Key;
1515
use list_query::ListLimitReached;
1616
pub use list_query::ListQuery;
17-
use pegboard_config::runner_protocol::proto::kv;
17+
use pegboard_runner_protocol::proto::kv;
1818
use prost::Message;
1919
use tokio::sync::Mutex;
2020
use utils::{validate_entries, validate_keys, TransactionExt};

0 commit comments

Comments
 (0)