Ebonilla/sy 749 opc ua read task cleanup and tests

.github/workflows/test.client.yaml

@@ -25,7 +25,7 @@ jobs:
         with:
           filters: |
             changed:
-                - "client/py/**"
+                - "client/**"
                 - "freighter/py/**"
                 - "synnax/**"
                 - "cesium/**"
@@ -194,3 +194,37 @@ jobs:
           password: ${{ secrets.GITHUB_TOKEN }}
         ports:
           - 9090:9090
+
+  test-cpp:
+    needs: [setup, build-server]
+    if: needs.setup.outputs.changed == 'true'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v3
+
+      - name: Setup Bazel
+        uses: bazel-contrib/setup-bazel@0.8.1
+        with:
+          bazelisk-cache: true
+          disk-cache: ${{ github.workflow }}
+          repository-cache: true
+
+      - name: Test
+        run: bazel test //client/...
+    services:
+      synnax:
+        image: ghcr.io/synnaxlabs/synnax:${{ github.sha }}
+        env:
+          SYNNAX_LISTEN: localhost:9090
+          SYNNAX_VERBOSE: true
+          SYNNAX_INSECURE: true
+          SYNNAX_MEM: true
+          SYNNAX_LICENSE_KEY: ${{ secrets.SYNNAX_LICENSE_KEY }}
+
+        credentials:
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+        ports:
+          - 9090:9090
+

cesium/internal/unary/writer.go

@@ -129,6 +129,8 @@
 	// closed stores whether the writer is closed. Operations like Write and Commit do not
 	// succeed on closed writers.
 	closed bool
+	// virtualAlignment tracks the alignment of the writer when persist is off.
+	virtualAlignment uint32
 }
 
 func (db *DB) OpenWriter(ctx context.Context, cfgs ...WriterConfig) (w *Writer, transfer controller.Transfer, err error) {
@@ -139,11 +141,13 @@
 	if err != nil {
 		return nil, transfer, err
 	}
-	w = &Writer{WriterConfig: cfg,
+	w = &Writer{
+		WriterConfig:     cfg,
 		Channel:          db.Channel,
 		idx:              db.index(),
 		decrementCounter: func() { db.entityCount.add(-1) },
 		wrapError:        db.wrapError,
+		virtualAlignment: 0,
 	}
 	gateCfg := controller.GateConfig{
 		TimeRange: cfg.controlTimeRange(),
@@ -206,23 +210,23 @@
 	if err = w.Channel.ValidateSeries(series); err != nil {
 		return 0, w.wrapError(err)
 	}
-	// ok signifies whether w is allowed to write.
 	dw, err := w.control.Authorize()
 	if err != nil {
-		return 0, err
+		return 0, w.wrapError(err)
 	}
-	a = telem.NewAlignmentPair(math.MaxUint32, uint32(w.len(dw.Writer)))
 	if w.Channel.IsIndex {
 		w.updateHwm(series)
 	}
 	if *w.Persist {
+		a = telem.NewAlignmentPair(math.MaxUint32, uint32(w.len(dw.Writer)))
 		_, err = dw.Write(series.Data)
+	} else {
+		a = telem.NewAlignmentPair(math.MaxUint32, w.virtualAlignment)
+		w.virtualAlignment += uint32(series.Len())
 	}
 	return a, w.wrapError(err)
 }
 
-func (w *Writer) SetPersist(persist bool) { w.Persist = config.Bool(persist) }
-
 func (w *Writer) SetAuthority(a control.Authority) controller.Transfer {
 	return w.control.SetAuthority(a)
 }

cesium/internal/virtual/db.go

@@ -17,9 +17,13 @@
 	"github.com/synnaxlabs/cesium/internal/meta"
 	"github.com/synnaxlabs/cesium/internal/version"
 	"github.com/synnaxlabs/x/binary"
+	"github.com/synnaxlabs/x/config"
+	"github.com/synnaxlabs/x/control"
 	"github.com/synnaxlabs/x/errors"
 	xfs "github.com/synnaxlabs/x/io/fs"
+	"github.com/synnaxlabs/x/override"
 	"github.com/synnaxlabs/x/telem"
+	"github.com/synnaxlabs/x/validate"
 	"sync"
 	"sync/atomic"
 )
@@ -58,8 +62,37 @@
 	Channel core.Channel
 }
 
-func Open(cfg Config) (db *DB, err error) {
-	c, err := controller.New[*controlEntity](controller.Config{Concurrency: cfg.Channel.Concurrency, Instrumentation: cfg.Instrumentation})
+var (
+	_             config.Config[Config] = Config{}
+	DefaultConfig                       = Config{}
+)
+
+// Validate implements config.Config.
+func (cfg Config) Validate() error {
+	v := validate.New("cesium.virtual")
+	validate.NotNil(v, "FS", cfg.FS)
+	return v.Error()
+}
+
+// Override implements config.Config.
+func (cfg Config) Override(other Config) Config {
+	cfg.FS = override.Nil(cfg.FS, other.FS)
+	if cfg.Channel.Key == 0 {
+		cfg.Channel = other.Channel
+	}
+	cfg.Instrumentation = override.Zero(cfg.Instrumentation, other.Instrumentation)
+	return cfg
+}
+
+func Open(configs ...Config) (db *DB, err error) {
+	cfg, err := config.New(DefaultConfig, configs...)
+	if err != nil {
+		return nil, err
+	}
+	c, err := controller.New[*controlEntity](controller.Config{
+		Concurrency:     control.Shared,
+		Instrumentation: cfg.Instrumentation,
+	})
 	if err != nil {
 		return nil, err
 	}

cesium/internal/virtual/write_test.go

@@ -7,6 +7,7 @@ import (
 	"github.com/synnaxlabs/cesium/internal/virtual"
 	"github.com/synnaxlabs/x/config"
 	"github.com/synnaxlabs/x/control"
+	xfs "github.com/synnaxlabs/x/io/fs"
 	"github.com/synnaxlabs/x/telem"
 	. "github.com/synnaxlabs/x/testutil"
 )
@@ -20,6 +21,7 @@ var _ = Describe("Write", func() {
 				DataType: telem.TimeStampT,
 				Virtual:  true,
 			},
+			FS: xfs.NewMem(),
 		}))
 	})
 	AfterEach(func() {
@@ -37,7 +39,7 @@ var _ = Describe("Write", func() {
 				Expect(t.Occurred()).To(BeTrue())
 				w2, t, err := db.OpenWriter(ctx, virtual.WriterConfig{
 					Start:             10 * telem.SecondTS,
-					Authority:         control.Absolute,
+					Authority:         control.Absolute - 1,
 					Subject:           control.Subject{Key: "bar"},
 					ErrOnUnauthorized: config.True(),
 				})

cesium/internal/virtual/writer.go

@@ -73,10 +73,7 @@ type Writer struct {
 	WriterConfig
 }
 
-func (db *DB) OpenWriter(
-	_ context.Context,
-	cfgs ...WriterConfig,
-) (w *Writer, transfer controller.Transfer, err error) {
+func (db *DB) OpenWriter(_ context.Context, cfgs ...WriterConfig) (w *Writer, transfer controller.Transfer, err error) {
 	if db.closed.Load() {
 		err = dbClosed
 		return nil, transfer, db.wrapError(err)

cesium/writer.go

@@ -91,25 +91,6 @@ func (w *Writer) Error() error {
 	return errors.New(unexpectedSteamClosure)
 }
 
-func (w *Writer) SetMode(mode WriterMode) bool {
-	if w.closed || w.hasAccumulatedErr {
-		return false
-	}
-	select {
-	case <-w.responses.Outlet():
-		w.hasAccumulatedErr = true
-		return false
-	case w.requests.Inlet() <- WriterRequest{Command: WriterSetMode, Config: WriterConfig{Mode: mode}}:
-	}
-	for res := range w.responses.Outlet() {
-		if res.Command == WriterSetMode {
-			return res.Ack
-		}
-	}
-	w.logger.DPanic(unexpectedSteamClosure)
-	return false
-}
-
 func (w *Writer) Close() (err error) {
 	if w.closed {
 		return nil

cesium/writer_behavior_test.go

@@ -1035,21 +1035,6 @@ var _ = Describe("Writer Behavior", func() {
 					Expect(frame.Series).To(HaveLen(0))
 					tsFrame := MustSucceed(db.Read(ctx, telem.TimeRangeMax, basic1Index))
 					Expect(tsFrame.Series).To(HaveLen(0))
-
-					By("Using SetMode to change the mode to persist")
-					Expect(w.SetMode(cesium.WriterPersistStream)).To(BeTrue())
-					ok = w.Write(cesium.NewFrame(
-						[]cesium.ChannelKey{basic1Index, basic1},
-						[]telem.Series{
-							telem.NewSecondsTSV(10, 11, 12, 13),
-							telem.NewSeriesV[int64](1, 2, 3, 4),
-						}),
-					)
-					Expect(ok).To(BeTrue())
-					end, ok = w.Commit()
-					Expect(ok).To(BeTrue())
-					Expect(end).To(Equal(13*telem.SecondTS + 1))
-
 					Expect(w.Close()).To(Succeed())
 				})
 			})

cesium/writer_stream.go

@@ -11,7 +11,6 @@
 
 import (
 	"context"
-	"fmt"
 	"github.com/synnaxlabs/cesium/internal/controller"
 	"github.com/synnaxlabs/cesium/internal/core"
 	"github.com/synnaxlabs/cesium/internal/index"
@@ -37,9 +36,6 @@
 	WriterError
 	// WriterSetAuthority represents a call to Writer.SetAuthority.
 	WriterSetAuthority
-	// WriterSetMode sets the operating WriterMode for the Writer. See the WriterMode
-	// documentation for more.
-	WriterSetMode
 )
 
 // WriterRequest is a request containing an arrow.Record to write to the DB.
@@ -126,8 +122,8 @@
 }
 
 func (w *streamWriter) process(ctx context.Context, req WriterRequest) {
-	if req.Command < WriterWrite || req.Command > WriterSetMode {
-		panic(fmt.Sprintf("invalid command %v", req.Command))
+	if req.Command < WriterWrite || req.Command > WriterSetAuthority {
+		return
 	}
 	if req.Command == WriterError {
 		w.seqNum++
@@ -141,12 +137,6 @@
 		w.sendRes(req, true, nil, 0)
 		return
 	}
-	if req.Command == WriterSetMode {
-		w.seqNum++
-		w.setMode(req.Config)
-		w.sendRes(req, true, nil, 0)
-		return
-	}
 	if w.err != nil {
 		w.seqNum++
 		w.sendRes(req, false, nil, 0)
@@ -216,16 +206,6 @@
 	}
 }
 
-func (w *streamWriter) setMode(cfg WriterConfig) {
-	persist := cfg.Mode < WriterStreamOnly
-	for _, idx := range w.internal {
-		for _, chW := range idx.internal {
-			chW.SetPersist(persist)
-		}
-	}
-	w.Mode = cfg.Mode
-}
-
 func (w *streamWriter) sendRes(req WriterRequest, ack bool, err error, end telem.TimeStamp) {
 	w.Out.Inlet() <- WriterResponse{
 		Command: req.Command,

client/cpp/.vscode/settings.json

@@ -1,5 +1,5 @@
 {
-    "files.associations": {
-        "memory": "cpp"
-    }
+  "files.associations": {
+    "memory": "cpp"
+  }
 }

client/cpp/auth/auth.h

@@ -10,9 +10,10 @@
 #pragma once
 
 #include <string>
+
+#include "client/cpp/errors/errors.h"
 #include "freighter/cpp/freighter.h"
 #include "synnax/pkg/api/grpc/v1/synnax/pkg/api/grpc/v1/auth.pb.h"
-#include "client/cpp/errors/errors.h"
 
 /// Auth meta data key. NOTE: This must be lowercase, GRPC will panic on capitalized
 /// or uppercase keys.

client/cpp/auth/auth_test.cpp

@@ -7,11 +7,13 @@
 // License, use of this software will be governed by the Apache License, Version 2.0,
 // included in the file licenses/APL.txt.
 
-#include <gtest/gtest.h>
-#include <memory>
 #include "client/cpp/auth/auth.h"
-#include "synnax/pkg/api/grpc/v1/synnax/pkg/api/grpc/v1/auth.pb.h"
+
+#include <memory>
+#include <gtest/gtest.h>
+
 #include "freighter/cpp/mock/mock.h"
+#include "synnax/pkg/api/grpc/v1/synnax/pkg/api/grpc/v1/auth.pb.h"
 
 /// @brief it should correctly authenticate with a Synnax cluster.
 TEST(TestAuth, testLoginHappyPath) {
@@ -54,61 +56,4 @@ TEST(TestAuth, testLoginInvalidCredentials) {
     auto [r, err] = mock_client.send("", v);
     EXPECT_TRUE(err) << err.message();
     EXPECT_TRUE(err.matches(synnax::INVALID_CREDENTIALS));
-}
-
-/// @brief it should retry authentication if the authentication token is invalid.
-TEST(TestAuth, testLoginRetry) {
-    auto res = api::v1::LoginResponse();
-    res.set_token("abc");
-    auto mock_login_client = std::make_unique<MockUnaryClient<
-        api::v1::LoginRequest,
-        api::v1::LoginResponse
-    >>(
-        std::vector<api::v1::LoginResponse>{res, res},
-        std::vector<freighter::Error>{freighter::NIL, freighter::NIL}
-    );
-    auto mw = std::make_shared<AuthMiddleware>(
-        std::move(mock_login_client),
-        "synnax",
-        "seldon",
-        3
-    );
-    auto mock_client = MockUnaryClient<int, int>{
-        {1, 1},
-        {freighter::Error(synnax::INVALID_TOKEN, ""), freighter::NIL}
-    };
-    mock_client.use(mw);
-    auto v = 1;
-    auto [r, err] = mock_client.send("", v);
-    EXPECT_FALSE(err) << err.message();
-    EXPECT_TRUE(err.matches(freighter::NIL));
-}
-
-/// @brief it should return an invalid token error if the maximum number of retries is exceeded.
-TEST(TestAuth, testExceedMaxRetries) {
-    auto res = api::v1::LoginResponse();
-    res.set_token("abc");
-    auto mock_login_client = std::make_unique<MockUnaryClient<
-        api::v1::LoginRequest,
-        api::v1::LoginResponse
-    >>(
-        std::vector<api::v1::LoginResponse>{res, res, res},
-        std::vector<freighter::Error>{freighter::NIL, freighter::NIL, freighter::NIL}
-    );
-    auto mw = std::make_shared<AuthMiddleware>(
-        std::move(mock_login_client),
-        "synnax",
-        "seldon",
-        2
-    );
-    auto invalid_token = freighter::Error(synnax::INVALID_TOKEN, "");
-    auto mock_client = MockUnaryClient<int, int>{
-        {1, 1, 1},
-        {invalid_token, invalid_token, invalid_token}
-    };
-    mock_client.use(mw);
-    auto v = 1;
-    auto [r, err] = mock_client.send("", v);
-    EXPECT_TRUE(err) << err.message();
-    EXPECT_TRUE(err.matches(synnax::INVALID_TOKEN));
 }