[cesium] - code cleanup

cesium/channel_test.go

@@ -534,6 +534,10 @@ var _ = Describe("Channel", Ordered, func() {
 					ch := MustSucceed(db.RetrieveChannel(ctx, key))
 					Expect(ch.Name).To(Equal("5"))
 				})
+				It("Should error if the channel is not found", func() {
+					key := GenerateChannelKey()
+					Expect(db.RenameChannel(ctx, key, "new_name")).To(HaveOccurredAs(cesium.ErrChannelNotFound))
+				})
 			})
 		})
 	}

cesium/delete_test.go

@@ -468,6 +468,14 @@ var _ = Describe("Delete", func() {
 						_, err = db.RetrieveChannel(ctx, index2)
 						Expect(err).To(BeNil())
 					})
+					It("Should be interrupted when there is an error in deleting index channels", func() {
+						i := MustSucceed(db.OpenIterator(cesium.IteratorConfig{Bounds: telem.TimeRangeMax, Channels: []core.ChannelKey{index1}}))
+						Expect(db.DeleteChannels([]cesium.ChannelKey{data1, index1})).To(MatchError(ContainSubstring("1 unclosed")))
+						Expect(i.Close()).To(Succeed())
+						Expect(fs.Exists(channelKeyToPath(data1))).To(BeFalse())
+						_, err := db.RetrieveChannels(ctx, data1)
+						Expect(err).To(HaveOccurredAs(cesium.ErrChannelNotFound))
+					})
 				})
 			})
 

cesium/internal/unary/delete.go

@@ -11,6 +11,7 @@
 
 import (
 	"context"
+	"fmt"
 	"github.com/cockroachdb/errors"
 	"github.com/google/uuid"
 	"github.com/synnaxlabs/x/control"
@@ -92,7 +93,9 @@
 			return offset, ts, err
 		}
 		if !approxStamp.Exact() {
-			panic("cannot find exact timestamp")
+			msg := fmt.Sprintf("cannot stamp deletion start %d offset from start %v", offset-1, domainStart)
+			db.L.DPanic(msg)
+			return offset, ts, errors.New(msg)
 		}
 		ts = approxStamp.Upper + 1
 	}
@@ -121,7 +124,9 @@
 			return offset, ts, err
 		}
 		if !approxStamp.Exact() {
-			panic("cannot find exact timestamp")
+			msg := fmt.Sprintf("cannot stamp deletion start %d offset from start %v", offset, domainStart)
+			db.L.DPanic(msg)
+			return offset, ts, errors.New(msg)
 		}
 		ts = approxStamp.Upper
 	}

cesium/internal/unary/iterator.go

@@ -52,9 +52,12 @@ type Iterator struct {
 
 const AutoSpan telem.TimeSpan = -1
 
+// SetBounds sets the iterator's bounds. The iterator is invalidated, and will not be
+// valid until a seeking call is made.
 func (i *Iterator) SetBounds(tr telem.TimeRange) {
 	i.bounds = tr
 	i.internal.SetBounds(tr)
+	i.seekReset(i.bounds.End)
 }
 
 func (i *Iterator) Bounds() telem.TimeRange { return i.bounds }

cesium/iterator.go

@@ -94,7 +94,8 @@ func (i *Iterator) Valid() bool {
 	return ok
 }
 
-// SetBounds implements Iterator.
+// SetBounds sets the iterator's bounds. The iterator is invalidated, and will not be
+// valid until a seeking call is made.
 func (i *Iterator) SetBounds(bounds telem.TimeRange) {
 	i.exec(IteratorRequest{Command: IterSetBounds, Bounds: bounds})
 }

cesium/iterator_test.go

@@ -91,8 +91,65 @@ var _ = Describe("Iterator Behavior", func() {
 
 					// Index1: 0  1  2 / _  4  _  _  7  _  9  /  10  _  12   _   _  15
 					// Data1: 10 11 12 / _ 14  _  _ 17  _ 19  /  20  _  22   _   _  25
-					// Index2: _  _  2   3  _  _  6  _  8  _  /  _  11  12  13   _  15
-					// Data2:  _  _  2   3  _  _  6  _  8  _  /  _  11  12  13   _  15
+					// Index2: _  _  2   3  _  _  6  _  8  /  _  _  11  12  13   _  15
+					// Data2:  _  _  2   3  _  _  6  _  8  /  _  _  11  12  13   _  15
+				})
+
+				Context("Basic", func() {
+					Specify("SeekLast", func() {
+						i := MustSucceed(db.OpenIterator(cesium.IteratorConfig{
+							Bounds:   telem.TimeRangeMax,
+							Channels: []cesium.ChannelKey{data1Key, data2Key},
+						}))
+						Expect(i.SeekLast()).To(BeTrue())
+						Expect(i.Prev(5 * telem.Second)).To(BeTrue())
+						f := i.Value()
+						Expect(f.Get(data1Key)[0].Data).To(EqualUnmarshal([]uint16{22, 25}))
+						Expect(f.Get(data2Key)[0].Data).To(EqualUnmarshal([]uint16{11, 12, 13, 15}))
+						Expect(i.Close()).To(Succeed())
+					})
+
+					Specify("SeekLE", func() {
+						i := MustSucceed(db.OpenIterator(cesium.IteratorConfig{
+							Bounds:   telem.TimeRangeMax,
+							Channels: []cesium.ChannelKey{data1Key, data2Key},
+						}))
+						Expect(i.SeekLE(4 * telem.SecondTS)).To(BeTrue())
+						Expect(i.Next(6 * telem.Second)).To(BeTrue())
+						f := i.Value()
+						Expect(f.Get(data1Key)[0].Data).To(EqualUnmarshal([]uint16{14, 17, 19}))
+						Expect(f.Get(data2Key)[0].Data).To(EqualUnmarshal([]uint16{6, 8}))
+						Expect(i.Close()).To(Succeed())
+					})
+
+					Specify("SeekGE", func() {
+						i := MustSucceed(db.OpenIterator(cesium.IteratorConfig{
+							Bounds:   telem.TimeRangeMax,
+							Channels: []cesium.ChannelKey{data1Key, data2Key},
+						}))
+						Expect(i.SeekGE(9 * telem.SecondTS)).To(BeTrue())
+						Expect(i.Next(3 * telem.Second)).To(BeTrue())
+						f := i.Value()
+						Expect(f.Get(data1Key)[0].Data).To(EqualUnmarshal([]uint16{19}))
+						Expect(f.Get(data1Key)[1].Data).To(EqualUnmarshal([]uint16{20}))
+						Expect(f.Get(data2Key)[0].Data).To(EqualUnmarshal([]uint16{11, 12, 13}))
+						Expect(i.Close()).To(Succeed())
+					})
+
+					Specify("SetBounds & Error", func() {
+						i := MustSucceed(db.OpenIterator(cesium.IteratorConfig{
+							Bounds:   telem.TimeRangeMax,
+							Channels: []cesium.ChannelKey{data1Key, data2Key},
+						}))
+						Expect(i.SeekGE(12 * telem.SecondTS)).To(BeTrue())
+						Expect(i.Next(3 * telem.Second)).To(BeTrue())
+						i.SetBounds((6 * telem.SecondTS).Range(9 * telem.SecondTS))
+						Expect(i.Valid()).To(BeFalse())
+						Expect(i.Error()).ToNot(HaveOccurred())
+						Expect(i.Close()).To(Succeed())
+						Expect(i.SeekFirst()).To(BeFalse())
+						Expect(i.Error()).To(MatchError(ContainSubstring("closed")))
+					})
 				})
 
 				Specify("With bound", func() {

cesium/options.go

@@ -62,12 +62,6 @@ func WithGC(config *GCConfig) Option {
 	}
 }
 
-func MemBacked() Option {
-	return func(o *options) {
-		o.fs = xfs.NewMem()
-	}
-}
-
 func WithInstrumentation(i alamos.Instrumentation) Option {
 	return func(o *options) {
 		o.Instrumentation = i

cesium/writer.go

@@ -76,6 +76,26 @@
 	return 0, false
 }
 
+// SetAuthority is synchronous
+func (w *Writer) SetAuthority(cfg WriterConfig) bool {
+	if w.closed || w.hasAccumulatedErr {
+		return false
+	}
+	select {
+	case <-w.responses.Outlet():
+		w.hasAccumulatedErr = true
+		return false
+	case w.requests.Inlet() <- WriterRequest{Config: cfg, Command: WriterSetAuthority}:
+	}
+	for res := range w.responses.Outlet() {
+		if res.Command == WriterSetAuthority {
+			return res.Ack
+		}
+	}
+	w.logger.DPanic(unexpectedSteamClosure)
+	return false
+}
+
 func (w *Writer) Error() error {
 	if w.closed {
 		return errWriterClosed

cesium/writer_test.go

@@ -1049,6 +1049,64 @@ var _ = Describe("Writer Behavior", func() {
 						Expect(w.Close()).To(Succeed())
 					})
 				})
+				Describe("Set authority", func() {
+					It("Should set the authority of writers", func() {
+						var (
+							key  = GenerateChannelKey()
+							key2 = GenerateChannelKey()
+						)
+						By("Creating a channel")
+						Expect(db.CreateChannel(
+							ctx,
+							cesium.Channel{Key: key, Rate: 1 * telem.Hz, DataType: telem.TimeStampT},
+							cesium.Channel{Key: key2, Virtual: true, DataType: telem.StringT},
+						)).To(Succeed())
+						w1 := MustSucceed(db.OpenWriter(ctx, cesium.WriterConfig{
+							Channels:       []cesium.ChannelKey{key, key2},
+							Start:          10 * telem.SecondTS,
+							Authorities:    []control.Authority{control.Authority(100), control.Authority(110)},
+							SendAuthErrors: config.True(),
+						}))
+
+						w2 := MustSucceed(db.OpenWriter(ctx, cesium.WriterConfig{
+							Channels:       []cesium.ChannelKey{key, key2},
+							Start:          10 * telem.SecondTS,
+							Authorities:    []control.Authority{control.Authority(110), control.Authority(100)},
+							SendAuthErrors: config.True(),
+						}))
+
+						w1.Write(cesium.NewFrame(
+							[]cesium.ChannelKey{key},
+							[]telem.Series{
+								telem.NewSeriesV[int64](1, 2, 3, 4),
+							}),
+						)
+						Expect(w1.Error()).To(HaveOccurredAs(control.Unauthorized))
+
+						Expect(w1.SetAuthority(cesium.WriterConfig{Channels: []cesium.ChannelKey{key, key2}, Authorities: []control.Authority{control.Absolute, control.Authority(0)}})).To(BeTrue())
+						Expect(w1.Error()).ToNot(HaveOccurred())
+
+						w2.Write(cesium.NewFrame(
+							[]cesium.ChannelKey{key},
+							[]telem.Series{
+								telem.NewSeriesV[int64](1, 3, 4),
+							},
+						))
+						Expect(w2.Error()).To(HaveOccurredAs(control.Unauthorized))
+
+						w1.Write(cesium.NewFrame(
+							[]cesium.ChannelKey{key2},
+							[]telem.Series{
+								{DataType: telem.StringT, Data: []byte("hehe")},
+							},
+						))
+
+						Expect(w1.Error()).To(HaveOccurredAs(control.Unauthorized))
+
+						Expect(w1.Close()).To(Succeed())
+						Expect(w2.Close()).To(Succeed())
+					})
+				})
 			})
 			Describe("Stream Only Mode", func() {
 				It("Should not persist data", func() {