joiningsource_test.go

// Copyright 2019 dfuse Platform Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package bstream

import (
	"context"
	"errors"
	"fmt"
	"testing"
	"time"

	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
	"github.com/streamingfast/dstore"

	"github.com/stretchr/testify/assert"
	"github.com/test-go/testify/require"
)

var errTestMock = errors.New("test failure")

func testHandler(failAt uint64) (HandlerFunc, chan *PreprocessedBlock) {
	out := make(chan *PreprocessedBlock, 100)
	return func(blk *pbbstream.Block, obj any) error {
		if blk.Number == failAt {
			return errTestMock
		}
		out <- &PreprocessedBlock{
			Block: blk,
			Obj:   obj,
		}
		return nil
	}, out
}

func TestJoiningSource_vanilla(t *testing.T) {
	joiningBlock := uint64(4)
	failingBlock := uint64(5)

	fileSF := NewTestSourceFactory()
	liveSF := NewTestSourceFactory()

	var liveSrc *TestSource

	handler, out := testHandler(failingBlock)
	liveSF.FromBlockNumFunc = func(num uint64, h Handler) Source {
		if num == joiningBlock {
			src := NewTestSource(h)
			liveSrc = src
			return src
		}
		return nil
	}

	joiningSource := NewJoiningSource(fileSF, liveSF, handler, 2, nil, false, zlog)
	go joiningSource.Run()

	fileSrc := <-fileSF.Created
	<-fileSrc.running // test fixture ready to push blocks
	assert.Equal(t, uint64(2), fileSrc.StartBlockNum)

	require.NoError(t, fileSrc.Push(TestBlock("00000002a", "00000001a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000003a", "00000002a"), nil))
	require.EqualError(t, fileSrc.Push(TestBlock("00000004a", "00000003a"), nil),
		stopSourceOnJoin.Error())

	<-fileSrc.Terminated() // previous error causes termination
	assert.Equal(t, 2, len(out))

	require.NotNil(t, liveSrc, "we should have joined to live source")

	require.NoError(t, liveSrc.Push(TestBlock("00000004a", "00000003a"), nil))
	assert.Equal(t, 3, len(out))

	require.EqualError(t, liveSrc.Push(TestBlock("00000005a", "00000004a"), nil),
		errTestMock.Error())

	<-liveSrc.Terminated()
	<-joiningSource.Terminated()
}

// TestJoiningSource_indexedArchiveRejoinsLiveAtBufferFloor checks that an
// index-filtered archive stream re-joins the live source at the live-buffer floor
// even when its filter matches nothing in the merged/live overlap and the merger
// is behind (no merged file above the overlap exists). This is the end-to-end
// behaviour guaranteed by FileSourceWithLiveBlockFloorGetter (firehose-core #109):
// the file source stops index-skipping at the floor, reads that bundle entirely,
// and emits an overlap block so the joining source can switch to live.
func TestJoiningSource_indexedArchiveRejoinsLiveAtBufferFloor(t *testing.T) {
	const liveFloor = uint64(250)

	// A historical bundle (where the request starts) and the bundle overlapping the
	// live buffer exist; everything above the overlap is missing (merger behind).
	// The request starts historical (block 1), so the index governs the climb toward
	// live: matching nothing, it would -- without the live floor -- skip straight past
	// the overlap bundle up to LastIndexedBlock and then wait forever for a not-yet
	// merged file, never emitting an overlap block and never re-joining live.
	mergedStore := dstore.NewMockStore(nil)
	mergedStore.SetFile(base(0), testBlocks(
		TestBlockWithNumbers("1a", "0a", 1, 0),
	))
	mergedStore.SetFile(base(200), testBlocks(
		TestBlockWithNumbers("250a", "249a", 250, 249),
		TestBlockWithNumbers("251a", "250a", 251, 250),
	))

	liveSF := NewTestSourceFactory()
	liveSF.LowestBlkNum = liveFloor
	joined := make(chan *TestSource, 1)
	liveSF.FromBlockNumFunc = func(num uint64, h Handler) Source {
		if num >= liveFloor {
			src := NewTestSource(h)
			joined <- src
			return src
		}
		return nil // below the floor: not a live block
	}

	fileSF := NewFileSourceFactory(mergedStore, dstore.NewMockStore(nil), zlog,
		FileSourceWithBlockIndexProvider(&TestBlockIndexProvider{Blocks: nil, LastIndexedBlock: 100000}),
		FileSourceWithLiveBlockFloorGetter(func() uint64 { return liveFloor }),
	)

	handler, out := testHandler(99999)
	js := NewJoiningSource(fileSF, liveSF, handler, 1, nil, false, zlog)
	go js.Run()
	defer js.Shutdown(nil)

	var liveSrc *TestSource
	select {
	case liveSrc = <-joined:
	case <-time.After(2 * time.Second):
		t.Fatal("indexed archive never re-joined the live source at the buffer floor")
	}

	// Live source now drives the stream forward; the historical block 1 was streamed
	// from archive before the re-join.
	require.NoError(t, liveSrc.Push(TestBlockWithNumbers("251a", "250a", 251, 250), nil))
	deadline := time.After(2 * time.Second)
	for {
		select {
		case b := <-out:
			if b.Block.Number >= liveFloor {
				require.Equal(t, uint64(251), b.Block.Number, "live block delivered after the re-join")
				return
			}
		case <-deadline:
			t.Fatal("expected a live block to be delivered after the re-join")
		}
	}
}

func TestJoiningSource_through_cursor(t *testing.T) {
	joiningBlock := uint64(6)
	failingBlock := uint64(9999)
	fileSF := NewTestSourceFactory()
	liveSF := NewTestSourceFactory()
	cursor := &Cursor{
		Step:      StepNew,
		Block:     NewBlockRefFromID("00000005"),
		HeadBlock: NewBlockRefFromID("00000005"),
		LIB:       NewBlockRefFromID("00000003"),
	}
	startBlock := uint64(3)

	var liveSrc *TestSource
	handler, out := testHandler(failingBlock)
	liveSF.ThroughCursorFunc = func(start uint64, cursor *Cursor, h Handler) Source {
		if start == joiningBlock {
			src := NewTestSource(h)
			src.Cursor = cursor
			src.StartBlockNum = start
			src.PassThroughCursor = true
			liveSrc = src
			return src
		}
		return nil
	}

	joiningSource := NewJoiningSource(fileSF, liveSF, handler, startBlock, cursor, true, zlog)
	go joiningSource.Run()

	fileSrc := <-fileSF.Created
	<-fileSrc.running // test fixture ready to push blocks
	assert.Equal(t, uint64(3), fileSrc.StartBlockNum)
	assert.Equal(t, cursor, fileSrc.Cursor)
	assert.True(t, fileSrc.PassThroughCursor)

	require.NoError(t, fileSrc.Push(TestBlock("00000003a", "00000002a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000004a", "00000003a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000005a", "00000004a"), nil))
	require.EqualError(t, fileSrc.Push(TestBlock("00000006a", "00000005a"), nil), stopSourceOnJoin.Error())

	<-fileSrc.Terminated()       // previous error causes termination
	assert.Equal(t, 3, len(out)) // 3, 4, 5 (6 will be sent by live)

	require.NotNil(t, liveSrc, "we should have joined to live source")
	assert.Equal(t, uint64(6), liveSrc.StartBlockNum)
	assert.Equal(t, cursor, liveSrc.Cursor)
	assert.True(t, liveSrc.PassThroughCursor)

	require.NoError(t, liveSrc.Push(TestBlock("00000006a", "00000005a"), nil))
	assert.Equal(t, 4, len(out))
}

func TestJoiningSource_skip_file_source(t *testing.T) {
	var fileSF ForkableSourceFactory //not used
	liveSF := NewTestSourceFactory()

	handler, out := testHandler(0)

	joiningSource := NewJoiningSource(fileSF, liveSF, handler, 2, nil, false, zlog)
	go joiningSource.Run()

	liveSrc := <-liveSF.Created
	<-liveSrc.running

	assert.Equal(t, uint64(2), liveSrc.StartBlockNum)

	require.NoError(t, liveSrc.Push(TestBlock("00000002a", "00000001a"), nil))
	require.NoError(t, liveSrc.Push(TestBlock("00000003a", "00000002a"), nil))

	assert.Len(t, out, 2)

	joiningSource.Shutdown(errTestMock)
	<-liveSrc.Terminated()       // previous error causes termination
	<-joiningSource.Terminated() // previous error causes termination
}

func TestJoiningSource_lowerLimitBackoff(t *testing.T) {
	fileSF := NewTestSourceFactory()
	liveSF := NewTestSourceFactory()

	liveSF.LowestBlkNum = 8
	joiningBlock := uint64(9)

	var liveSrc *TestSource
	liveSourceFactoryCalls := 0

	handler, out := testHandler(0)
	liveSF.FromBlockNumFunc = func(num uint64, h Handler) Source {
		liveSourceFactoryCalls++
		if num == joiningBlock {
			src := NewTestSource(h)
			liveSrc = src
			return src
		}
		return nil
	}

	joiningSource := NewJoiningSource(fileSF, liveSF, handler, 1, nil, false, zlog)
	go joiningSource.Run()

	fileSrc := <-fileSF.Created
	<-fileSrc.running
	assert.Equal(t, uint64(1), fileSrc.StartBlockNum)

	require.NoError(t, fileSrc.Push(TestBlock("00000001a", "00000000a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000002a", "00000001a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000003a", "00000002a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000004a", "00000003a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000005a", "00000004a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000006a", "00000005a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000007a", "00000006a"), nil))
	require.NoError(t, fileSrc.Push(TestBlock("00000008a", "00000007a"), nil))

	require.EqualError(t, fileSrc.Push(TestBlock("00000009a", "00000008a"), nil),
		stopSourceOnJoin.Error())

	<-fileSrc.Terminated() // previous error causes termination
	assert.Equal(t, 8, len(out))

	require.NotNil(t, liveSrc, "we should have joined to live source")
	require.NoError(t, liveSrc.Push(TestBlock("00000009a", "00000008a"), nil))
	assert.Equal(t, 9, len(out))

	assert.Equal(t, 3, liveSourceFactoryCalls)

}

// testLiveKnower is a live source factory that also answers what its buffer holds, the
// way the forkable hub does.
type testLiveKnower struct {
	*TestSourceFactory
	lowest uint64
	head   uint64
	blocks map[string]*pbbstream.Block

	headAfterCatchUp uint64
	catchUpAt        time.Time
}

func (t *testLiveKnower) LowestBlockNum() uint64 { return t.lowest }

// HeadNum reports headAfterCatchUp once catchUpAt has passed, standing in for a live
// source that is behind and catching up while the check waits on it.
func (t *testLiveKnower) HeadNum() uint64 {
	if t.headAfterCatchUp != 0 && time.Now().After(t.catchUpAt) {
		return t.headAfterCatchUp
	}
	return t.head
}
func (t *testLiveKnower) GetBlockByHash(id string) *pbbstream.Block {
	return t.blocks[id]
}

// testForkedKnower is a file source factory that also answers what the forked-blocks
// store holds.
type testForkedKnower struct {
	*TestSourceFactory
	forkedBlocks map[string]bool // "<num>-<id suffix>"
	err          error
}

func (t *testForkedKnower) HasForkedBlock(ctx context.Context, idSuffix string, blockNum uint64) (bool, error) {
	if t.err != nil {
		return false, t.err
	}
	return t.forkedBlocks[fmt.Sprintf("%d-%s", blockNum, idSuffix)], nil
}

func TestJoiningSourceCheckCursorResolvable(t *testing.T) {
	// the wait on a lagging live source is bounded by this; keep the test quick
	previousTimeout := CursorHeadWaitTimeout
	CursorHeadWaitTimeout = 300 * time.Millisecond
	defer func() { CursorHeadWaitTimeout = previousTimeout }()

	knownID := "00000000000000000000000000000000000000000000000000000000000000aa"
	unknownID := "00000000000000000000000000000000000000000000000000000000000000bb"

	cursorAt := func(id string, num uint64) *Cursor {
		return &Cursor{
			Step:      StepNew,
			Block:     NewBlockRef(id, num),
			HeadBlock: NewBlockRef(id, num),
			LIB:       NewBlockRef(knownID, 100),
		}
	}

	tests := []struct {
		name             string
		cursor           *Cursor
		lowest           uint64
		head             uint64
		headAfterCatchUp uint64
		catchUpAfter     time.Duration
		forked           map[string]bool
		forkedErr        error
		expectErrror     bool
		expectAboveHead  bool
	}{
		{
			name:   "no cursor",
			cursor: nil,
			lowest: 100, head: 200,
		},
		{
			name:   "live buffer holds the cursor block",
			cursor: cursorAt(knownID, 150),
			lowest: 100, head: 200,
		},
		{
			name:   "cursor block below the live buffer, left to the file source",
			cursor: cursorAt(unknownID, 50),
			lowest: 100, head: 200,
		},
		{
			name:   "cursor block above the live head, reached while waiting",
			cursor: cursorAt(knownID, 250),
			lowest: 100, head: 200,
			headAfterCatchUp: 260,
			catchUpAfter:     50 * time.Millisecond,
		},
		{
			name:   "cursor block above the live head, never reached",
			cursor: cursorAt(unknownID, 250),
			lowest: 100, head: 200,
			expectAboveHead: true,
		},
		{
			name:   "cursor block reached while waiting, and unknown there",
			cursor: cursorAt(unknownID, 250),
			lowest: 100, head: 200,
			headAfterCatchUp: 260,
			catchUpAfter:     50 * time.Millisecond,
			expectErrror:     true,
		},
		{
			name:   "live buffer not ready, left to the file source",
			cursor: cursorAt(unknownID, 150),
			lowest: 0, head: 0,
		},
		{
			name:   "unknown inside the live buffer, forked blocks hold it",
			cursor: cursorAt(unknownID, 150),
			lowest: 100, head: 200,
			forked: map[string]bool{"150-" + TruncateBlockID(unknownID): true},
		},
		{
			name:   "forked blocks hold that ID at another height, which is another block",
			cursor: cursorAt(unknownID, 150),
			lowest: 100, head: 200,
			forked:       map[string]bool{"149-" + TruncateBlockID(unknownID): true},
			expectErrror: true,
		},
		{
			name:      "unknown inside the live buffer, forked blocks store fails",
			cursor:    cursorAt(unknownID, 150),
			lowest:    100,
			head:      200,
			forkedErr: errTestMock,
		},
		{
			name:   "unknown inside the live buffer and nowhere else",
			cursor: cursorAt(unknownID, 150),
			lowest: 100, head: 200,
			expectErrror: true,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			live := &testLiveKnower{
				TestSourceFactory: NewTestSourceFactory(),
				lowest:            test.lowest,
				head:              test.head,
				blocks: map[string]*pbbstream.Block{
					knownID: {Id: knownID, Number: 150},
				},
				headAfterCatchUp: test.headAfterCatchUp,
				catchUpAt:        time.Now().Add(test.catchUpAfter),
			}
			file := &testForkedKnower{
				TestSourceFactory: NewTestSourceFactory(),
				forkedBlocks:      test.forked,
				err:               test.forkedErr,
			}

			s := NewJoiningSource(file, live, nil, 100, test.cursor, false, zlog)

			err := s.checkCursorResolvable()
			if test.expectAboveHead {
				require.Error(t, err)
				assert.ErrorIs(t, err, ErrCursorAboveHead)
				return
			}
			if test.expectErrror {
				require.Error(t, err)
				assert.ErrorIs(t, err, ErrResolveCursor)
				return
			}
			assert.NoError(t, err)
		})
	}
}