Add subscribe offset underflow test
@@ -564,3 +564,89 @@ func TestSubscribeOffsetOverflowEmptyStream(t *testing.T) {
t.Fatal("Did not receive expected message")
}
}
+
+// Ensure when the subscribe offset is less than the oldest log offset, the
+// offset is set to the oldest offset.
+func TestSubscribeOffsetUnderflow(t *testing.T) {
+ defer cleanupStorage(t)
+
+ // Use a central NATS server.
+ ns := natsdTest.RunDefaultServer()
+ defer ns.Shutdown()
+
+ // Configure server.
+ s1Config := getTestConfig("a", true, 5050)
+ // Set these to force deletion so we can get an underflow.
+ s1Config.Log.SegmentMaxBytes = 1
+ s1Config.Log.RetentionMaxBytes = 1
+ s1Config.BatchMaxMessages = 1
+ s1 := runServerWithConfig(t, s1Config)
+ defer s1.Stop()
+
+ // Wait for server to elect itself leader.
+ getMetadataLeader(t, 10*time.Second, s1)
+
+ client, err := liftbridge.Connect([]string{"localhost:5050"})
+ require.NoError(t, err)
+ defer client.Close()
+
+ // Create stream.
+ stream := liftbridge.StreamInfo{
+ Name: "foo",
+ Subject: "foo",
+ ReplicationFactor: 1,
+ }
+ err = client.CreateStream(context.Background(), stream)
+ require.NoError(t, err)
+
+ nc, err := nats.GetDefaultOptions().Connect()
+ require.NoError(t, err)
+ defer nc.Close()
+
+ // Subscribe to acks.
+ acks := "acks"
+ num := 2
+ acked := 0
+ gotAcks := make(chan struct{})
+ _, err = nc.Subscribe(acks, func(m *nats.Msg) {
+ _, err := liftbridge.UnmarshalAck(m.Data)
+ require.NoError(t, err)
+ acked++
+ if acked == num {
+ close(gotAcks)
+ }
+ })
+ require.NoError(t, err)
+ nc.Flush()
+
+ // Publish some messages.
+ for i := 0; i < num; i++ {
+ err = nc.Publish(stream.Subject, liftbridge.NewMessage([]byte("hello"),
+ liftbridge.MessageOptions{AckInbox: acks}))
+ require.NoError(t, err)
+ }
+
+ // Wait for acks.
+ select {
+ case <-gotAcks:
+ case <-time.After(5 * time.Second):
+ t.Fatal("Did not receive expected acks")
+ }
+
+ // Subscribe with underflowed offset. This should set the offset to 1.
+ gotMsg := make(chan struct{})
+ ctx, cancel := context.WithCancel(context.Background())
+ client.Subscribe(ctx, stream.Subject, stream.Name, func(msg *proto.Message, err error) {
+ require.NoError(t, err)
+ require.Equal(t, int64(1), msg.Offset)
+ close(gotMsg)
+ cancel()
+ })
+
+ // Wait to get the new message.
+ select {
+ case <-gotMsg:
+ case <-time.After(5 * time.Second):
+ t.Fatal("Did not receive expected message")
+ }
+}
