Skip to content

Commit c2a439c

Browse files
emilbon99emilbon99
authored
SY-4048: Synnax v0.54.4 (#2211)
1 parent 92e8aef commit c2a439c

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

89 files changed

+3660
-929
lines changed

‎aspen/internal/cluster/config.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"github.com/synnaxlabs/x/encoding"
2121
"github.com/synnaxlabs/x/encoding/gob"
2222
"github.com/synnaxlabs/x/encoding/json"
23+
"github.com/synnaxlabs/x/encoding/msgpack"
2324
"github.com/synnaxlabs/x/kv"
2425
"github.com/synnaxlabs/x/override"
2526
"github.com/synnaxlabs/x/validate"
@@ -98,9 +99,9 @@ var (
9899
StorageKey: []byte("aspen.cluster"),
99100
Gossip: gossip.DefaultConfig,
100101
StorageFlushInterval: 1 * time.Second,
101-
// This used to be implemented by a gob codec, but we want to switch to msgpack.
102-
// Instead, we will use a fallback codec that tries msgpack to decode first, then gob.
103-
Codec: encoding.NewDecodeFallbackCodec(json.Codec, gob.Codec),
102+
// Encodes as JSON, decodes with fallback: JSON -> MsgPack -> Gob.
103+
// MsgPack was the primary codec from v0.39 to v0.53, Gob before that.
104+
Codec: encoding.NewDecodeFallbackCodec(json.Codec, msgpack.Codec, gob.Codec),
104105
}
105106
FastConfig = DefaultConfig.Override(Config{
106107
Pledge: pledge.FastConfig,

‎aspen/internal/cluster/join_test.go‎

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818
"github.com/synnaxlabs/aspen/internal/node"
1919
"github.com/synnaxlabs/freighter/mock"
2020
"github.com/synnaxlabs/x/address"
21+
"github.com/synnaxlabs/x/encoding/gob"
2122
"github.com/synnaxlabs/x/encoding/msgpack"
2223
"github.com/synnaxlabs/x/kv/memkv"
2324
"github.com/synnaxlabs/x/signal"
@@ -151,6 +152,157 @@ var _ = Describe("Open", func() {
151152
Expect(kvDB.Close()).To(Succeed())
152153
})
153154

155+
It("Should recover state written by msgpack (v0.39 to v0.53 upgrade)", func(ctx SpecContext) {
156+
gossipT1 := gossipNet.UnaryServer("")
157+
pledgeT1 := pledgeNet.UnaryServer(gossipT1.Address)
158+
clusterOne := MustSucceed(cluster.Open(
159+
ctx,
160+
cluster.Config{
161+
HostAddress: gossipT1.Address,
162+
Pledge: pledge.Config{
163+
Peers: []address.Address{},
164+
TransportClient: pledgeNet.UnaryClient(),
165+
TransportServer: pledgeT1,
166+
},
167+
Gossip: gossip.Config{
168+
TransportClient: gossipNet.UnaryClient(),
169+
TransportServer: gossipT1,
170+
Interval: 100 * time.Millisecond,
171+
},
172+
},
173+
))
174+
175+
kvDB := memkv.New()
176+
gossipT2 := gossipNet.UnaryServer("")
177+
pledgeT2 := pledgeNet.UnaryServer(gossipT2.Address)
178+
storageKey := []byte("msgpack-upgrade-test")
179+
180+
// Simulate a v0.39-v0.53 server that wrote state as msgpack.
181+
oldConfig := cluster.Config{
182+
HostAddress: gossipT2.Address,
183+
Pledge: pledge.Config{
184+
Peers: []address.Address{gossipT1.Address},
185+
TransportClient: pledgeNet.UnaryClient(),
186+
TransportServer: pledgeT2,
187+
},
188+
Gossip: gossip.Config{
189+
TransportClient: gossipNet.UnaryClient(),
190+
TransportServer: gossipT2,
191+
Interval: 100 * time.Millisecond,
192+
},
193+
StorageKey: storageKey,
194+
Storage: kvDB,
195+
StorageFlushInterval: cluster.FlushOnEvery,
196+
Codec: msgpack.Codec,
197+
}
198+
clusterTwo := MustSucceed(cluster.Open(ctx, oldConfig))
199+
Expect(clusterTwo.Host().Key).To(Equal(node.Key(2)))
200+
Expect(clusterTwo.Close()).To(Succeed())
201+
202+
// Reopen with the default codec (JSON primary, msgpack+gob fallback),
203+
// simulating an upgrade to v0.54+.
204+
gossipT3 := gossipNet.UnaryServer(gossipT2.Address)
205+
pledgeT3 := pledgeNet.UnaryServer(gossipT3.Address)
206+
upgradedConfig := cluster.Config{
207+
HostAddress: gossipT3.Address,
208+
Pledge: pledge.Config{
209+
Peers: []address.Address{gossipT1.Address},
210+
TransportClient: pledgeNet.UnaryClient(),
211+
TransportServer: pledgeT3,
212+
},
213+
Gossip: gossip.Config{
214+
TransportClient: gossipNet.UnaryClient(),
215+
TransportServer: gossipT3,
216+
Interval: 100 * time.Millisecond,
217+
},
218+
StorageKey: storageKey,
219+
Storage: kvDB,
220+
StorageFlushInterval: cluster.FlushOnEvery,
221+
}
222+
clusterTwoAgain := MustSucceed(cluster.Open(ctx, upgradedConfig))
223+
Expect(clusterTwoAgain.Host().Key).To(Equal(node.Key(2)))
224+
Expect(clusterTwoAgain.Nodes()).To(HaveLen(2))
225+
226+
Expect(clusterOne.Close()).To(Succeed())
227+
Expect(clusterTwoAgain.Close()).To(Succeed())
228+
Expect(kvDB.Close()).To(Succeed())
229+
})
230+
231+
It("Should recover state written by gob (pre-v0.39 upgrade)", func(ctx SpecContext) {
232+
gossipT1 := gossipNet.UnaryServer("")
233+
pledgeT1 := pledgeNet.UnaryServer(gossipT1.Address)
234+
clusterOne := MustSucceed(cluster.Open(
235+
ctx,
236+
cluster.Config{
237+
HostAddress: gossipT1.Address,
238+
Pledge: pledge.Config{
239+
Peers: []address.Address{},
240+
TransportClient: pledgeNet.UnaryClient(),
241+
TransportServer: pledgeT1,
242+
},
243+
Gossip: gossip.Config{
244+
TransportClient: gossipNet.UnaryClient(),
245+
TransportServer: gossipT1,
246+
Interval: 100 * time.Millisecond,
247+
},
248+
},
249+
))
250+
251+
kvDB := memkv.New()
252+
gossipT2 := gossipNet.UnaryServer("")
253+
pledgeT2 := pledgeNet.UnaryServer(gossipT2.Address)
254+
storageKey := []byte("gob-upgrade-test")
255+
256+
// Simulate a pre-v0.39 server that wrote state as gob.
257+
oldConfig := cluster.Config{
258+
HostAddress: gossipT2.Address,
259+
Pledge: pledge.Config{
260+
Peers: []address.Address{gossipT1.Address},
261+
TransportClient: pledgeNet.UnaryClient(),
262+
TransportServer: pledgeT2,
263+
},
264+
Gossip: gossip.Config{
265+
TransportClient: gossipNet.UnaryClient(),
266+
TransportServer: gossipT2,
267+
Interval: 100 * time.Millisecond,
268+
},
269+
StorageKey: storageKey,
270+
Storage: kvDB,
271+
StorageFlushInterval: cluster.FlushOnEvery,
272+
Codec: gob.Codec,
273+
}
274+
clusterTwo := MustSucceed(cluster.Open(ctx, oldConfig))
275+
Expect(clusterTwo.Host().Key).To(Equal(node.Key(2)))
276+
Expect(clusterTwo.Close()).To(Succeed())
277+
278+
// Reopen with the default codec, simulating an upgrade to v0.54+.
279+
gossipT3 := gossipNet.UnaryServer(gossipT2.Address)
280+
pledgeT3 := pledgeNet.UnaryServer(gossipT3.Address)
281+
upgradedConfig := cluster.Config{
282+
HostAddress: gossipT3.Address,
283+
Pledge: pledge.Config{
284+
Peers: []address.Address{gossipT1.Address},
285+
TransportClient: pledgeNet.UnaryClient(),
286+
TransportServer: pledgeT3,
287+
},
288+
Gossip: gossip.Config{
289+
TransportClient: gossipNet.UnaryClient(),
290+
TransportServer: gossipT3,
291+
Interval: 100 * time.Millisecond,
292+
},
293+
StorageKey: storageKey,
294+
Storage: kvDB,
295+
StorageFlushInterval: cluster.FlushOnEvery,
296+
}
297+
clusterTwoAgain := MustSucceed(cluster.Open(ctx, upgradedConfig))
298+
Expect(clusterTwoAgain.Host().Key).To(Equal(node.Key(2)))
299+
Expect(clusterTwoAgain.Nodes()).To(HaveLen(2))
300+
301+
Expect(clusterOne.Close()).To(Succeed())
302+
Expect(clusterTwoAgain.Close()).To(Succeed())
303+
Expect(kvDB.Close()).To(Succeed())
304+
})
305+
154306
})
155307

156308
})

‎client/py/examples/opcua/server.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -356,6 +356,12 @@ def __init__(
356356
self.user_manager = user_manager
357357
self.max_sessions = max_sessions
358358

359+
def start(self) -> None:
360+
super().start()
361+
# Allow OPCUA Server time to startup
362+
# so server doesn't reject the Core for trying to connect too many times
363+
sy.sleep(5)
364+
359365
async def _run_server(self) -> None:
360366
await run_server(
361367
self.endpoint,

‎client/py/examples/simulators/press.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,7 @@ def _create_channels(self) -> None:
116116

117117
def _run_loop(self) -> None:
118118
self.log("Starting simulation loop...")
119-
loop = sy.Loop(sy.Rate.HZ * 100)
119+
loop = sy.Loop(sy.Rate.HZ * 50)
120120
loop_count = 0
121121

122122
state = {
@@ -166,10 +166,10 @@ def _run_loop(self) -> None:
166166
break
167167

168168
if state["press_vlv_state"] == 1:
169-
state["press_pt"] += 0.2
169+
state["press_pt"] += 0.4
170170

171171
if state["vent_vlv_state"] == 1:
172-
state["press_pt"] -= 0.2
172+
state["press_pt"] -= 0.4
173173

174174
if state["press_pt"] < 0:
175175
state["press_pt"] = 0

‎client/py/examples/simulators/thermal.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@ class ThermalSimDAQ(SimDAQ):
2020
end_cmd_channel = "end_thermal_test_cmd"
2121

2222
AMBIENT_TEMP = 25.0
23-
HEAT_RATE = 0.5
24-
COOL_RATE = 0.2
23+
HEAT_RATE = 1.0
24+
COOL_RATE = 0.4
2525
NOISE = 0.1
2626

2727
def _create_channels(self) -> None:
@@ -105,7 +105,7 @@ def _create_channels(self) -> None:
105105

106106
def _run_loop(self) -> None:
107107
self.log("Starting simulation loop...")
108-
loop = sy.Loop(sy.Rate.HZ * 100)
108+
loop = sy.Loop(sy.Rate.HZ * 50)
109109
loop_count = 0
110110

111111
state = {

‎client/py/pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[project]
22
name = "synnax"
3-
version = "0.54.1"
3+
version = "0.54.2"
44
description = "Synnax Client Library"
55
keywords = ["Synnax", "Synnax Python Client"]
66
authors = [{ name = "Emiliano Bonilla", email = "ebonilla@synnaxlabs.com" }]

‎client/ts/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@synnaxlabs/client",
3-
"version": "0.54.1",
3+
"version": "0.54.2",
44
"description": "The Synnax Client Library",
55
"keywords": [
66
"synnax",

‎client/ts/src/client.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
// License, use of this software will be governed by the Apache License, Version 2.0,
88
// included in the file licenses/APL.txt.
99

10-
import { breaker, TimeSpan, TimeStamp, URL } from "@synnaxlabs/x";
10+
import { breaker, TimeSpan, TimeStamp, URL, zod } from "@synnaxlabs/x";
1111
import { z } from "zod";
1212

1313
import { access } from "@/access";
@@ -109,7 +109,7 @@ export default class Synnax extends framer.Client {
109109
* the client from polling the cluster for connectivity information.
110110
*/
111111
constructor(params: SynnaxParams) {
112-
const parsedParams = synnaxParamsZ.parse(params);
112+
const parsedParams = zod.parse(synnaxParamsZ, params);
113113
const {
114114
host,
115115
port,
@@ -196,7 +196,7 @@ export const checkConnection = async (params: CheckConnectionParams) =>
196196

197197
export const newConnectionChecker = (params: CheckConnectionParams) => {
198198
const { host, port, secure, name, retry } = params;
199-
const retryConfig = breaker.breakerConfigZ.optional().parse(retry);
199+
const retryConfig = zod.parse(breaker.breakerConfigZ.optional(), retry);
200200
const url = new URL({ host, port: Number(port) });
201201
const transport = new Transport(url, retryConfig, secure);
202202
return new connection.Checker(transport.unary, undefined, __VERSION__, name);

‎client/ts/src/framer/writer.ts‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,14 @@
88
// included in the file licenses/APL.txt.
99

1010
import { EOF, type Stream, type WebSocketClient } from "@synnaxlabs/freighter";
11-
import { control, type CrudeSeries, errors, TimeSpan, TimeStamp } from "@synnaxlabs/x";
11+
import {
12+
control,
13+
type CrudeSeries,
14+
errors,
15+
TimeSpan,
16+
TimeStamp,
17+
zod,
18+
} from "@synnaxlabs/x";
1219
import { z } from "zod";
1320

1421
import { channel } from "@/channel";
@@ -203,7 +210,7 @@ export class Writer {
203210
client: WebSocketClient,
204211
config: WriterConfig,
205212
): Promise<Writer> {
206-
const cfg = writerConfigZ.parse(config);
213+
const cfg = zod.parse(writerConfigZ, config);
207214
const adapter = await WriteAdapter.open(retriever, cfg.channels);
208215
if (cfg.useHighPerformanceCodec)
209216
client = client.withCodec(new WSWriterCodec(adapter.codec));

‎client/ts/src/rack/rack.spec.ts‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,8 @@
77
// License, use of this software will be governed by the Apache License, Version 2.0,
88
// included in the file licenses/APL.txt.
99

10-
import { TimeStamp } from "@synnaxlabs/x";
10+
import { TimeStamp, zod } from "@synnaxlabs/x";
1111
import { describe, expect, it } from "vitest";
12-
import { ZodError } from "zod";
1312

1413
import { type rack } from "@/rack";
1514
import { createTestClient } from "@/testutil/client";
@@ -24,7 +23,7 @@ describe("Rack", () => {
2423
});
2524
it("should return an error if the rack doesn't have a name", async () => {
2625
// @ts-expect-error - Testing for error
27-
await expect(client.racks.create({})).rejects.toThrow(ZodError);
26+
await expect(client.racks.create({})).rejects.toThrow(zod.ParseError);
2827
});
2928
it("should create a rack with a custom status", async () => {
3029
const customStatus: rack.Status = {

0 commit comments

Comments
 (0)