From 3b3f4024a991cc966635bb7298a5e2b2a4672629 Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Wed, 25 Sep 2024 12:11:44 -0700
Subject: [PATCH 01/10] [client/py] - fixed race condition in set sizing

---
 .../py/examples/control/manual-abort/seq.py   | 34 +++++++++++++++++++
 client/py/synnax/control/controller.py        | 21 +++++++++---
 2 files changed, 50 insertions(+), 5 deletions(-)
 create mode 100644 client/py/examples/control/manual-abort/seq.py

diff --git a/client/py/examples/control/manual-abort/seq.py b/client/py/examples/control/manual-abort/seq.py
new file mode 100644
index 0000000000..417ab08902
--- /dev/null
+++ b/client/py/examples/control/manual-abort/seq.py
@@ -0,0 +1,34 @@
+#  Copyright 2024 Synnax Labs, Inc.
+#
+#  Use of this software is governed by the Business Source License included in the file
+#  licenses/BSL.txt.
+#
+#  As of the Change Date specified in that file, in accordance with the Business Source
+#  License, use of this software will be governed by the Apache License, Version 2.0,
+#  included in the file licenses/APL.txt.
+
+import synnax as sy
+
+client = sy.Synnax(
+    host="10.0.0.210",
+    port=9090,
+    username="synnax",
+    password="seldon"
+)
+
+ABORT_BUTTON = "test_bool"
+DATA = "USB-6008_ai_0"
+
+abort_ch = client.channels.retrieve(ABORT_BUTTON)
+
+def print_state(auto):
+    print(auto.state)
+    return abort_ch.key in auto.state
+
+with client.control.acquire(
+    name="manual-abort",
+    write=[ABORT_BUTTON],
+    read=[ABORT_BUTTON, DATA],
+) as auto:
+    auto.set(ABORT_BUTTON, False)
+    auto.wait_until(print_state)
diff --git a/client/py/synnax/control/controller.py b/client/py/synnax/control/controller.py
index 0a135ca503..34c6edec2a 100644
--- a/client/py/synnax/control/controller.py
+++ b/client/py/synnax/control/controller.py
@@ -10,7 +10,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from threading import Event
+from threading import Event, Lock
 from typing import Any, Protocol, overload
 from asyncio import create_task, Future
 
@@ -281,13 +281,13 @@ def _internal_wait_until(
             raise ValueError("First argument to wait_until must be a callable.")
         processor = WaitUntil(cond, reverse)
         try:
-            self._receiver.processors.add(processor)
+            self._receiver.add_processor(processor)
             timeout_seconds = (
                 TimeSpan.from_seconds(timeout).seconds if timeout else None
             )
             ok = processor.event.wait(timeout=timeout_seconds)
         finally:
-            self._receiver.processors.remove(processor)
+            self._receiver.remove_processor(processor)
         if processor.exc:
             raise processor.exc
         return ok
@@ -458,6 +458,7 @@ class _Receiver(AsyncThread):
     client: framer.Client
     streamer: framer.AsyncStreamer
     processors: set[Processor]
+    processor_lock: Lock
     retriever: ChannelRetriever
     controller: Controller
     startup_ack: Event
@@ -475,12 +476,22 @@ def __init__(
         self.client = client
         self.state = dict()
         self.controller = controller
+        self.processor_lock = Lock()
         self.startup_ack = Event()
         self.processors = set()
 
+    def add_processor(self, processor: Processor):
+        with self.processor_lock:
+            self.processors.add(processor)
+
+    def remove_processor(self, processor: Processor):
+        with self.processor_lock:
+            self.processors.remove(processor)
+
     def _process(self):
-        for p in self.processors:
-            p.process(self.controller)
+        with self.processor_lock:
+            for p in self.processors:
+                p.process(self.controller)
 
     async def _listen_for_close(self):
         await self.shutdown_future

From 71955ec5b3a8c61e3f4f0007a953d2e8ca23b784 Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Wed, 25 Sep 2024 13:01:53 -0700
Subject: [PATCH 02/10] [client] - adjusted streamer bootup sequence

---
 client/cpp/framer/streamer.cpp                   |  4 +++-
 client/py/synnax/framer/streamer.py              |  7 +++++++
 client/py/tests/test_control.py                  |  1 -
 client/py/tests/test_framer.py                   |  8 ++++++++
 client/py/tests/test_timing.py                   |  1 -
 client/ts/src/framer/streamer.spec.ts            |  3 +++
 client/ts/src/framer/streamer.ts                 |  2 ++
 synnax/pkg/api/framer.go                         |  5 ++++-
 synnax/pkg/distribution/framer/relay/relay.go    | 15 ++++++++++-----
 synnax/pkg/distribution/framer/relay/streamer.go | 14 +++++++-------
 synnax/pkg/distribution/framer/service.go        |  1 +
 11 files changed, 45 insertions(+), 16 deletions(-)

diff --git a/client/cpp/framer/streamer.cpp b/client/cpp/framer/streamer.cpp
index 8e812e4313..0edf6d1d70 100644
--- a/client/cpp/framer/streamer.cpp
+++ b/client/cpp/framer/streamer.cpp
@@ -28,7 +28,9 @@ std::pair<Streamer, freighter::Error> FrameClient::openStreamer(
     auto req = api::v1::FrameStreamerRequest();
     config.toProto(req);
     auto exc2 = s->send(req);
-    return {Streamer(std::move(s)), exc2};
+    if (err) {Streamer(std::move(s)), exc2};
+    auto [_, resExc] = s->receive();
+    return {Streamer(std::move(s)), resExc};
 }
 
 Streamer::Streamer(std::unique_ptr<StreamerStream> s) : stream(std::move(s)) {
diff --git a/client/py/synnax/framer/streamer.py b/client/py/synnax/framer/streamer.py
index af9b6b7c22..33d3b735de 100644
--- a/client/py/synnax/framer/streamer.py
+++ b/client/py/synnax/framer/streamer.py
@@ -70,6 +70,10 @@ def __init__(
 
     def __open(self):
         self._stream.send(_Request(keys=self._adapter.keys))
+        _, exc = self._stream.receive()
+        if exc is not None:
+            raise exc
+
 
     @overload
     def read(self, timeout: float | int | TimeSpan) -> Frame | None:
@@ -201,6 +205,9 @@ def __init__(
     async def _open(self):
         self._stream = await self._client.stream(_ENDPOINT, _Request, _Response)
         await self._stream.send(_Request(keys=self._adapter.keys))
+        _, exc = await self._stream.receive()
+        if exc is not None:
+            raise exc
 
     @property
     def received(self) -> bool:
diff --git a/client/py/tests/test_control.py b/client/py/tests/test_control.py
index f8465c4c53..ec5d8d9a30 100644
--- a/client/py/tests/test_control.py
+++ b/client/py/tests/test_control.py
@@ -365,7 +365,6 @@ def test_controller_channel_not_found(self, client: sy.Synnax):
                 v = auto[press_en.key]
                 assert v is None
 
-    @pytest.mark.focus
     def test_controller_set_authority_mechanisms(self, client: sy.Synnax):
         press_end_cmd_time, press_en_cmd, press_en, daq_time = create_valve_set(client)
 
diff --git a/client/py/tests/test_framer.py b/client/py/tests/test_framer.py
index 6ca3ebcc5b..e643c6270b 100644
--- a/client/py/tests/test_framer.py
+++ b/client/py/tests/test_framer.py
@@ -403,6 +403,14 @@ def test_open_streamer_no_channels(self, client: sy.Synnax):
         with client.open_streamer([]):
             pass
 
+
+    @pytest.mark.focus
+    def test_open_streamer_channel_not_found(self, client: sy.Synnax):
+        """Should throw an exception when a streamer is opened with an unknown channel"""
+        with pytest.raises(sy.NotFoundError):
+            with client.open_streamer([123]):
+                pass
+
     def test_update_channels(self, channel: sy.Channel, client: sy.Synnax):
         """Should update the list of channels to stream"""
         with client.open_streamer([]) as s:
diff --git a/client/py/tests/test_timing.py b/client/py/tests/test_timing.py
index 67781052fc..fdf1fef25a 100644
--- a/client/py/tests/test_timing.py
+++ b/client/py/tests/test_timing.py
@@ -39,7 +39,6 @@ def test_sleep(self):
 
         assert sum(accumulated_precise) < sum(accumulated_standard)
 
-    @pytest.mark.focus
     def test_sleep_rate(self):
         """Should sleep correctly based on a rate argument
         """
diff --git a/client/ts/src/framer/streamer.spec.ts b/client/ts/src/framer/streamer.spec.ts
index d062629c4a..420ac92dbd 100644
--- a/client/ts/src/framer/streamer.spec.ts
+++ b/client/ts/src/framer/streamer.spec.ts
@@ -47,4 +47,7 @@ describe("Streamer", () => {
   it("should not throw an error when the streamer is opened with zero channels", async () => {
     await expect(client.openStreamer([])).resolves.not.toThrow();
   });
+  it("should throw an error when the streamer is opened with a channel that does not exist", async () => {
+    await expect(client.openStreamer([5678])).rejects.toThrow("not found");
+  });
 });
diff --git a/client/ts/src/framer/streamer.ts b/client/ts/src/framer/streamer.ts
index 69df5c1364..88f02eeb0e 100644
--- a/client/ts/src/framer/streamer.ts
+++ b/client/ts/src/framer/streamer.ts
@@ -55,6 +55,8 @@ export class Streamer implements AsyncIterator<Frame>, AsyncIterable<Frame> {
     const stream = await client.stream(ENDPOINT, reqZ, resZ);
     const streamer = new Streamer(stream, adapter);
     stream.send({ keys: adapter.keys });
+    const [, err] = await stream.receive();
+    if (err != null) throw err;
     return streamer;
   }
 
diff --git a/synnax/pkg/api/framer.go b/synnax/pkg/api/framer.go
index 1c8e47954c..4289cb9803 100644
--- a/synnax/pkg/api/framer.go
+++ b/synnax/pkg/api/framer.go
@@ -178,7 +178,10 @@ func (s *FrameService) openStreamer(ctx context.Context, stream StreamerStream)
 		return nil, err
 	}
 	reader, err := s.Internal.NewStreamer(ctx, framer.StreamerConfig{Keys: req.Keys})
-	return reader, err
+	if err != nil {
+		return nil, err
+	}
+	return reader, stream.Send(framer.StreamerResponse{})
 }
 
 type FrameWriterConfig struct {
diff --git a/synnax/pkg/distribution/framer/relay/relay.go b/synnax/pkg/distribution/framer/relay/relay.go
index 2b5fc61ac6..d7a9da3f7a 100644
--- a/synnax/pkg/distribution/framer/relay/relay.go
+++ b/synnax/pkg/distribution/framer/relay/relay.go
@@ -11,6 +11,7 @@ package relay
 
 import (
 	"fmt"
+	"github.com/synnaxlabs/synnax/pkg/distribution/channel"
 	"time"
 
 	"github.com/synnaxlabs/alamos"
@@ -27,10 +28,11 @@ import (
 
 type Config struct {
 	alamos.Instrumentation
-	Transport    Transport
-	HostResolver core.HostResolver
-	TS           *ts.DB
-	FreeWrites   confluence.Outlet[Response]
+	Transport     Transport
+	HostResolver  core.HostResolver
+	TS            *ts.DB
+	FreeWrites    confluence.Outlet[Response]
+	ChannelReader channel.Readable
 }
 
 var (
@@ -45,6 +47,7 @@ func (c Config) Override(other Config) Config {
 	c.HostResolver = override.Nil(c.HostResolver, other.HostResolver)
 	c.TS = override.Nil(c.TS, other.TS)
 	c.FreeWrites = override.Nil(c.FreeWrites, other.FreeWrites)
+	c.ChannelReader = override.Nil(c.ChannelReader, other.ChannelReader)
 	return c
 }
 
@@ -55,10 +58,12 @@ func (c Config) Validate() error {
 	validate.NotNil(v, "HostProvider", c.HostResolver)
 	validate.NotNil(v, "TS", c.TS)
 	validate.NotNil(v, "FreeWrites", c.FreeWrites)
+	validate.NotNil(v, "ChannelReader", c.ChannelReader)
 	return v.Error()
 }
 
 type Relay struct {
+	cfg     Config
 	ins     alamos.Instrumentation
 	delta   *confluence.DynamicDeltaMultiplier[Response]
 	demands confluence.Inlet[demand]
@@ -75,7 +80,7 @@ func Open(configs ...Config) (*Relay, error) {
 		return nil, err
 	}
 
-	r := &Relay{ins: cfg.Instrumentation}
+	r := &Relay{cfg: cfg, ins: cfg.Instrumentation}
 
 	tpr := newTapper(cfg)
 	demands := confluence.NewStream[demand](defaultBuffer)
diff --git a/synnax/pkg/distribution/framer/relay/streamer.go b/synnax/pkg/distribution/framer/relay/streamer.go
index a7fbf27077..202d5b1a61 100644
--- a/synnax/pkg/distribution/framer/relay/streamer.go
+++ b/synnax/pkg/distribution/framer/relay/streamer.go
@@ -31,13 +31,13 @@ type streamer struct {
 
 type StreamerConfig = Request
 
-func (r *Relay) NewStreamer(_ context.Context, cfg StreamerConfig) (Streamer, error) {
-	return &streamer{
-		keys:    lo.Uniq(cfg.Keys),
-		addr:    address.Rand(),
-		demands: r.demands,
-		relay:   r,
-	}, nil
+func (r *Relay) NewStreamer(ctx context.Context, cfg StreamerConfig) (Streamer, error) {
+	keys := lo.Uniq(cfg.Keys)
+	// Check that all keys exist.
+	if err := r.cfg.ChannelReader.NewRetrieve().WhereKeys(keys...).Exec(ctx, nil); err != nil {
+		return nil, err
+	}
+	return &streamer{keys: keys, addr: address.Rand(), demands: r.demands, relay: r}, nil
 }
 
 // Flow implements confluence.Flow.
diff --git a/synnax/pkg/distribution/framer/service.go b/synnax/pkg/distribution/framer/service.go
index a1427b0616..cb14d0266f 100644
--- a/synnax/pkg/distribution/framer/service.go
+++ b/synnax/pkg/distribution/framer/service.go
@@ -116,6 +116,7 @@ func Open(configs ...Config) (*Service, error) {
 	freeWrites := confluence.NewStream[relay.Response](25)
 	s.Relay, err = relay.Open(relay.Config{
 		Instrumentation: cfg.Instrumentation.Child("Relay"),
+		ChannelReader:   cfg.ChannelReader,
 		TS:              cfg.TS,
 		HostResolver:    cfg.HostResolver,
 		Transport:       cfg.Transport.Relay(),

From 0d0c90c5ddcad8c5c9518c32942ef9b8930babc9 Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Wed, 25 Sep 2024 18:54:19 -0700
Subject: [PATCH 03/10] [client/py] - adjusted manual abort example

---
 .../py/examples/control/manual-abort/seq.py   | 34 -------------------
 1 file changed, 34 deletions(-)
 delete mode 100644 client/py/examples/control/manual-abort/seq.py

diff --git a/client/py/examples/control/manual-abort/seq.py b/client/py/examples/control/manual-abort/seq.py
deleted file mode 100644
index 417ab08902..0000000000
--- a/client/py/examples/control/manual-abort/seq.py
+++ /dev/null
@@ -1,34 +0,0 @@
-#  Copyright 2024 Synnax Labs, Inc.
-#
-#  Use of this software is governed by the Business Source License included in the file
-#  licenses/BSL.txt.
-#
-#  As of the Change Date specified in that file, in accordance with the Business Source
-#  License, use of this software will be governed by the Apache License, Version 2.0,
-#  included in the file licenses/APL.txt.
-
-import synnax as sy
-
-client = sy.Synnax(
-    host="10.0.0.210",
-    port=9090,
-    username="synnax",
-    password="seldon"
-)
-
-ABORT_BUTTON = "test_bool"
-DATA = "USB-6008_ai_0"
-
-abort_ch = client.channels.retrieve(ABORT_BUTTON)
-
-def print_state(auto):
-    print(auto.state)
-    return abort_ch.key in auto.state
-
-with client.control.acquire(
-    name="manual-abort",
-    write=[ABORT_BUTTON],
-    read=[ABORT_BUTTON, DATA],
-) as auto:
-    auto.set(ABORT_BUTTON, False)
-    auto.wait_until(print_state)

From ce92d40a4e287d15a2a774f47b892bd0ebf63c13 Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Wed, 25 Sep 2024 19:00:39 -0700
Subject: [PATCH 04/10] [framer] - adjusted relay test

---
 synnax/pkg/distribution/framer/relay/relay_suite_test.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/synnax/pkg/distribution/framer/relay/relay_suite_test.go b/synnax/pkg/distribution/framer/relay/relay_suite_test.go
index b13848ba89..1921235108 100644
--- a/synnax/pkg/distribution/framer/relay/relay_suite_test.go
+++ b/synnax/pkg/distribution/framer/relay/relay_suite_test.go
@@ -75,6 +75,7 @@ func provision(n int) (*mock.CoreBuilder, map[core.NodeKey]serviceContainer) {
 			TS:              c.Storage.TS,
 			Transport:       relayNet.New(c.Config.AdvertiseAddress),
 			HostResolver:    c.Cluster,
+			ChannelReader:   container.channel,
 			FreeWrites:      freeWrites,
 		}))
 		container.writer = MustSucceed(writer.OpenService(writer.ServiceConfig{

From 65646e71358205134ae39ffab71d917f5bb436a4 Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Wed, 25 Sep 2024 19:01:56 -0700
Subject: [PATCH 05/10] [ops] - bumped version

---
 client/py/pyproject.toml   | 2 +-
 client/ts/package.json     | 2 +-
 console/package.json       | 2 +-
 synnax/pkg/version/VERSION | 2 +-
 4 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/client/py/pyproject.toml b/client/py/pyproject.toml
index 5ae03d973f..3e3caa450b 100644
--- a/client/py/pyproject.toml
+++ b/client/py/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "synnax"
-version = "0.29.0"
+version = "0.29.1"
 description = "Synnax Client Library"
 keywords = ["Synnax", "Synnax Python Client"]
 authors = ["emiliano bonilla <emilbon99@gmail.com>"]
diff --git a/client/ts/package.json b/client/ts/package.json
index 1171033f15..a05df51811 100644
--- a/client/ts/package.json
+++ b/client/ts/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/client",
-  "version": "0.29.0",
+  "version": "0.29.1",
   "description": "The Synnax Client Library",
   "keywords": [
     "synnax",
diff --git a/console/package.json b/console/package.json
index 406f81f4db..3a7da3d59f 100644
--- a/console/package.json
+++ b/console/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@synnaxlabs/console",
   "private": true,
-  "version": "0.29.1",
+  "version": "0.29.2",
   "type": "module",
   "scripts": {
     "dev": "tauri dev",
diff --git a/synnax/pkg/version/VERSION b/synnax/pkg/version/VERSION
index ae6dd4e203..25939d35c7 100644
--- a/synnax/pkg/version/VERSION
+++ b/synnax/pkg/version/VERSION
@@ -1 +1 @@
-0.29.0
+0.29.1

From 16a2db004b38d167be00e830b5786de02b3d8bfd Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Wed, 25 Sep 2024 19:08:46 -0700
Subject: [PATCH 06/10] [client/cpp] - fixed undefined variable

---
 client/cpp/framer/streamer.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client/cpp/framer/streamer.cpp b/client/cpp/framer/streamer.cpp
index 0edf6d1d70..cbc500b97d 100644
--- a/client/cpp/framer/streamer.cpp
+++ b/client/cpp/framer/streamer.cpp
@@ -28,7 +28,7 @@ std::pair<Streamer, freighter::Error> FrameClient::openStreamer(
     auto req = api::v1::FrameStreamerRequest();
     config.toProto(req);
     auto exc2 = s->send(req);
-    if (err) {Streamer(std::move(s)), exc2};
+    if (exc2) {Streamer(std::move(s)), exc2};
     auto [_, resExc] = s->receive();
     return {Streamer(std::move(s)), resExc};
 }

From 561c7c63e64899362f61c6145181fdf4782dd808 Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Wed, 25 Sep 2024 19:13:38 -0700
Subject: [PATCH 07/10] [client/cpp] - added missing return statement

---
 client/cpp/framer/streamer.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client/cpp/framer/streamer.cpp b/client/cpp/framer/streamer.cpp
index cbc500b97d..7e0cc0ca4b 100644
--- a/client/cpp/framer/streamer.cpp
+++ b/client/cpp/framer/streamer.cpp
@@ -28,7 +28,7 @@ std::pair<Streamer, freighter::Error> FrameClient::openStreamer(
     auto req = api::v1::FrameStreamerRequest();
     config.toProto(req);
     auto exc2 = s->send(req);
-    if (exc2) {Streamer(std::move(s)), exc2};
+    if (exc2) return {Streamer(std::move(s)), exc2};
     auto [_, resExc] = s->receive();
     return {Streamer(std::move(s)), resExc};
 }

From 95e64dda8eef6f4282085790c08e759dcb8a9d48 Mon Sep 17 00:00:00 2001
From: Emiliano Bonilla <56323762+emilbon99@users.noreply.github.com>
Date: Thu, 26 Sep 2024 08:20:05 -0700
Subject: [PATCH 08/10] [synnax] - added tests to check for channels not found
 in streamer relay

---
 synnax/pkg/distribution/framer/relay/relay_test.go | 14 ++++++++++++++
 synnax/pkg/distribution/framer/relay/streamer.go   |  5 ++++-
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/synnax/pkg/distribution/framer/relay/relay_test.go b/synnax/pkg/distribution/framer/relay/relay_test.go
index 3ccfc40961..47ea95acfd 100644
--- a/synnax/pkg/distribution/framer/relay/relay_test.go
+++ b/synnax/pkg/distribution/framer/relay/relay_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/synnaxlabs/x/confluence"
 	"github.com/synnaxlabs/x/errors"
 	xio "github.com/synnaxlabs/x/io"
+	"github.com/synnaxlabs/x/query"
 	"github.com/synnaxlabs/x/signal"
 	"github.com/synnaxlabs/x/telem"
 	. "github.com/synnaxlabs/x/testutil"
@@ -99,6 +100,19 @@ var _ = Describe("Relay", func() {
 			})
 		}
 	})
+	Describe("Errors", func() {
+		It("Should raise an error if a channel is not found", func() {
+			builder, services := provision(1)
+			defer func() {
+				Expect(builder.Close()).To(Succeed())
+			}()
+			svc := services[1]
+			_, err := svc.relay.NewStreamer(context.TODO(), relay.StreamerConfig{
+				Keys: []channel.Key{12345},
+			})
+			Expect(err).To(HaveOccurredAs(query.NotFound))
+		})
+	})
 })
 
 func newChannelSet() []channel.Channel {
diff --git a/synnax/pkg/distribution/framer/relay/streamer.go b/synnax/pkg/distribution/framer/relay/streamer.go
index 202d5b1a61..575811f8dc 100644
--- a/synnax/pkg/distribution/framer/relay/streamer.go
+++ b/synnax/pkg/distribution/framer/relay/streamer.go
@@ -11,6 +11,7 @@ package relay
 
 import (
 	"context"
+
 	"github.com/samber/lo"
 	"github.com/synnaxlabs/synnax/pkg/distribution/channel"
 	"github.com/synnaxlabs/x/address"
@@ -34,7 +35,9 @@ type StreamerConfig = Request
 func (r *Relay) NewStreamer(ctx context.Context, cfg StreamerConfig) (Streamer, error) {
 	keys := lo.Uniq(cfg.Keys)
 	// Check that all keys exist.
-	if err := r.cfg.ChannelReader.NewRetrieve().WhereKeys(keys...).Exec(ctx, nil); err != nil {
+	if err := r.cfg.ChannelReader.
+		NewRetrieve().
+		WhereKeys(keys...).Exec(ctx, nil); err != nil {
 		return nil, err
 	}
 	return &streamer{keys: keys, addr: address.Rand(), demands: r.demands, relay: r}, nil

From 481a08fffd5464a1a3860a921e19726934da6be4 Mon Sep 17 00:00:00 2001
From: Elham Islam <eislam@umich.edu>
Date: Thu, 26 Sep 2024 09:24:29 -0700
Subject: [PATCH 09/10] bump major version

---
 alamos/py/pyproject.toml                      |  2 +-
 alamos/ts/package.json                        |  2 +-
 client/py/examples/control/daq/nominal_seq.py | 64 +++++++++++++++++++
 client/py/pyproject.toml                      |  2 +-
 client/ts/package.json                        |  2 +-
 console/package.json                          |  2 +-
 drift/package.json                            |  2 +-
 freighter/py/pyproject.toml                   |  2 +-
 freighter/ts/package.json                     |  2 +-
 package.json                                  |  2 +-
 pluto/package.json                            |  2 +-
 synnax/pkg/version/VERSION                    |  2 +-
 x/ts/package.json                             |  2 +-
 13 files changed, 76 insertions(+), 12 deletions(-)
 create mode 100644 client/py/examples/control/daq/nominal_seq.py

diff --git a/alamos/py/pyproject.toml b/alamos/py/pyproject.toml
index e6c2ba1af7..1b2a45a700 100644
--- a/alamos/py/pyproject.toml
+++ b/alamos/py/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "alamos"
-version = "0.29.0"
+version = "0.30.0"
 description = ""
 authors = ["Emiliano Bonilla <ebonilla@synnaxlabs.com>"]
 readme = "README.md"
diff --git a/alamos/ts/package.json b/alamos/ts/package.json
index c69521908d..e8622d27d4 100644
--- a/alamos/ts/package.json
+++ b/alamos/ts/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/alamos",
-  "version": "0.29.0",
+  "version": "0.30.0",
   "type": "module",
   "description": "Distributed instrumentation for Synnax",
   "repository": "https://github.com/synnaxlabs/synnax/tree/main/freighter/ts",
diff --git a/client/py/examples/control/daq/nominal_seq.py b/client/py/examples/control/daq/nominal_seq.py
new file mode 100644
index 0000000000..6a10b62bf2
--- /dev/null
+++ b/client/py/examples/control/daq/nominal_seq.py
@@ -0,0 +1,64 @@
+#  Copyright 2024 Synnax Labs, Inc.
+#
+#  Use of this software is governed by the Business Source License included in the file
+#  licenses/BSL.txt.
+#
+#  As of the Change Date specified in that file, in accordance with the Business Source
+#  License, use of this software will be governed by the Apache License, Version 2.0,
+#  included in the file licenses/APL.txt.
+
+import synnax as sy
+import time
+
+# We've logged in via the CLI, so there's no need to provide credentials here. See
+# https://docs.synnaxlabs.com/reference/python-client/get-started for more information.
+client = sy.Synnax(
+    host="10.0.0.210",
+    port=9090,
+    username="synnax",
+    password="seldon",
+    secure=False
+)
+# Define the control channel names
+NI_CHANNEL = "USB-6008_ai_0"
+BOOL_CHANNEL = "test_bool"
+
+bool_channel = client.channels.retrieve(BOOL_CHANNEL)
+
+print("BOOL", bool_channel)
+
+def print_out(auto):
+    print(auto.state)
+    return bool_channel.key in auto.state
+
+# Open a control sequence under a context manager, so that the control is released when
+# the block exits
+with client.control.acquire(
+    name="Press Sequence",
+    write=[NI_CHANNEL, BOOL_CHANNEL],
+    read =[NI_CHANNEL, BOOL_CHANNEL],
+    write_authorities=[200],
+) as ctrl:
+    ctrl.set({
+        BOOL_CHANNEL: True,
+    })
+    ctrl.wait_until(print_out)
+    # ctrl.wait_until(
+    #     lambda c: c[NI_CHANNEL] > 0.5,
+    #     timeout=20 * sy.TimeSpan.SECOND,
+    # )
+    # print(ctrl[NI_CHANNEL])
+
+
+# with client.control.acquire(
+#     name="Press Sequence",
+#     read =[ BOOL_CHANNEL],
+#     write_authorities=[200],
+# ) as ctrl:
+#     ctrl.sleep(1)
+#     ctrl.set({
+#         BOOL_CHANNEL: True,
+#     })
+#     ctrl.wait_until_defined(BOOL_CHANNEL, timeout=20 * sy.TimeSpan.SECOND)
+#     print("TEST_BOOL: ", ctrl[BOOL_CHANNEL])
+  
\ No newline at end of file
diff --git a/client/py/pyproject.toml b/client/py/pyproject.toml
index 3e3caa450b..28c0f3e4d3 100644
--- a/client/py/pyproject.toml
+++ b/client/py/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "synnax"
-version = "0.29.1"
+version = "0.30.0"
 description = "Synnax Client Library"
 keywords = ["Synnax", "Synnax Python Client"]
 authors = ["emiliano bonilla <emilbon99@gmail.com>"]
diff --git a/client/ts/package.json b/client/ts/package.json
index a05df51811..39d50805e1 100644
--- a/client/ts/package.json
+++ b/client/ts/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/client",
-  "version": "0.29.1",
+  "version": "0.30.0",
   "description": "The Synnax Client Library",
   "keywords": [
     "synnax",
diff --git a/console/package.json b/console/package.json
index 3a7da3d59f..df135fc685 100644
--- a/console/package.json
+++ b/console/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@synnaxlabs/console",
   "private": true,
-  "version": "0.29.2",
+  "version": "0.30.0",
   "type": "module",
   "scripts": {
     "dev": "tauri dev",
diff --git a/drift/package.json b/drift/package.json
index 92b62e83ca..f09c321cd2 100644
--- a/drift/package.json
+++ b/drift/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/drift",
-  "version": "0.29.0",
+  "version": "0.30.0",
   "description": "State synchronization and Redux state synchronization for Tauri Apps",
   "repository": "https://github.com/synnaxlabs/synnax/tree/main/drift",
   "type": "module",
diff --git a/freighter/py/pyproject.toml b/freighter/py/pyproject.toml
index 450c679c10..152aa9c71c 100644
--- a/freighter/py/pyproject.toml
+++ b/freighter/py/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "synnax-freighter"
-version = "0.29.0"
+version = "0.30.0"
 description = ""
 authors = ["emiliano bonilla <emilbon99@gmail.com>"]
 packages = [
diff --git a/freighter/ts/package.json b/freighter/ts/package.json
index 0ca5e37d30..fc2012c4df 100644
--- a/freighter/ts/package.json
+++ b/freighter/ts/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/freighter",
-  "version": "0.29.0",
+  "version": "0.30.0",
   "type": "module",
   "description": "a modular transport abstraction",
   "repository": "https://github.com/synnaxlabs/synnax/tree/main/freighter/ts",
diff --git a/package.json b/package.json
index a80df35a67..30a96b6283 100644
--- a/package.json
+++ b/package.json
@@ -1,5 +1,5 @@
 {
-  "version": "0.29.0",
+  "version": "0.30.0",
   "private": true,
   "scripts": {
     "build": "npx turbo build --cache-dir=./.turbo-cache",
diff --git a/pluto/package.json b/pluto/package.json
index 99e0485a8c..5a628fa696 100644
--- a/pluto/package.json
+++ b/pluto/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/pluto",
-  "version": "0.29.0",
+  "version": "0.30.0",
   "type": "module",
   "scripts": {
     "build": "tsc --noEmit && vite build",
diff --git a/synnax/pkg/version/VERSION b/synnax/pkg/version/VERSION
index 25939d35c7..c25c8e5b74 100644
--- a/synnax/pkg/version/VERSION
+++ b/synnax/pkg/version/VERSION
@@ -1 +1 @@
-0.29.1
+0.30.0
diff --git a/x/ts/package.json b/x/ts/package.json
index 667edb9876..43d491a206 100644
--- a/x/ts/package.json
+++ b/x/ts/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/x",
-  "version": "0.29.0",
+  "version": "0.30.0",
   "type": "module",
   "description": "Common Utilities for Synnax Labs",
   "repository": "https://github.com/synnaxlabs/synnax/tree/main/x/go",

From d7e50393768b7d10b4289cee61531739bba1874a Mon Sep 17 00:00:00 2001
From: Elham Islam <eislam@umich.edu>
Date: Thu, 26 Sep 2024 09:25:23 -0700
Subject: [PATCH 10/10] remove file accidentally pushed

---
 client/py/examples/control/daq/nominal_seq.py | 64 -------------------
 1 file changed, 64 deletions(-)
 delete mode 100644 client/py/examples/control/daq/nominal_seq.py

diff --git a/client/py/examples/control/daq/nominal_seq.py b/client/py/examples/control/daq/nominal_seq.py
deleted file mode 100644
index 6a10b62bf2..0000000000
--- a/client/py/examples/control/daq/nominal_seq.py
+++ /dev/null
@@ -1,64 +0,0 @@
-#  Copyright 2024 Synnax Labs, Inc.
-#
-#  Use of this software is governed by the Business Source License included in the file
-#  licenses/BSL.txt.
-#
-#  As of the Change Date specified in that file, in accordance with the Business Source
-#  License, use of this software will be governed by the Apache License, Version 2.0,
-#  included in the file licenses/APL.txt.
-
-import synnax as sy
-import time
-
-# We've logged in via the CLI, so there's no need to provide credentials here. See
-# https://docs.synnaxlabs.com/reference/python-client/get-started for more information.
-client = sy.Synnax(
-    host="10.0.0.210",
-    port=9090,
-    username="synnax",
-    password="seldon",
-    secure=False
-)
-# Define the control channel names
-NI_CHANNEL = "USB-6008_ai_0"
-BOOL_CHANNEL = "test_bool"
-
-bool_channel = client.channels.retrieve(BOOL_CHANNEL)
-
-print("BOOL", bool_channel)
-
-def print_out(auto):
-    print(auto.state)
-    return bool_channel.key in auto.state
-
-# Open a control sequence under a context manager, so that the control is released when
-# the block exits
-with client.control.acquire(
-    name="Press Sequence",
-    write=[NI_CHANNEL, BOOL_CHANNEL],
-    read =[NI_CHANNEL, BOOL_CHANNEL],
-    write_authorities=[200],
-) as ctrl:
-    ctrl.set({
-        BOOL_CHANNEL: True,
-    })
-    ctrl.wait_until(print_out)
-    # ctrl.wait_until(
-    #     lambda c: c[NI_CHANNEL] > 0.5,
-    #     timeout=20 * sy.TimeSpan.SECOND,
-    # )
-    # print(ctrl[NI_CHANNEL])
-
-
-# with client.control.acquire(
-#     name="Press Sequence",
-#     read =[ BOOL_CHANNEL],
-#     write_authorities=[200],
-# ) as ctrl:
-#     ctrl.sleep(1)
-#     ctrl.set({
-#         BOOL_CHANNEL: True,
-#     })
-#     ctrl.wait_until_defined(BOOL_CHANNEL, timeout=20 * sy.TimeSpan.SECOND)
-#     print("TEST_BOOL: ", ctrl[BOOL_CHANNEL])
-  
\ No newline at end of file