Merge pull request #19 from streamingfast/block_any

v1.0.0 firecore binary, new block format (based on anypb)

.github/workflows/docker.yml

@@ -0,0 +1,78 @@
+name: Build docker image
+
+on:
+  push:
+    tags:
+      - "v*"
+    branches:
+      - "block_any"
+  workflow_dispatch:
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+
+jobs:
+  build:
+    runs-on: ubuntu-20.04
+
+    permissions:
+      contents: read
+      packages: write
+
+    strategy:
+      matrix:
+        go-version: [ 1.20.x ]
+
+    outputs:
+      tags: ${{ steps.meta.outputs.tags }}
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Log in to the Container registry
+        uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Get repo name
+        id: extract_repo_name
+        shell: bash
+        run: |
+          echo "REPO_NAME=$(basename ${{ github.repository }})" >> $GITHUB_ENV
+
+      - name: Generate docker tags/labels from github build context
+        id: meta
+        uses: docker/metadata-action@v4
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=ref,event=tag
+            type=sha,prefix=,enable=true
+            type=raw,enable=${{ github.ref == 'refs/heads/develop' }},value=develop
+          flavor: |
+            latest=${{ startsWith(github.ref, 'refs/tags/') }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v4
+        with:
+          context: .
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+
+
+  slack-notifications:
+    if: ${{ !startsWith(github.ref, 'refs/tags/') && github.event_name != 'workflow_dispatch' }}
+    needs: [ build ]
+    runs-on: ubuntu-20.04
+    steps:
+      - name: Slack notification
+        env:
+          SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
+        uses: Ilshidur/[email protected]
+        with:
+          args: |
+            :done: *${{ github.repository }}* Success building docker image from ${{ github.ref_type }} _${{ github.ref_name }}_ (${{ github.actor }}) :sparkling_heart: ```${{ join(needs.build.outputs.tags, ' ') }}```

.gitignore

@@ -1,3 +1,5 @@
 .idea
 /build
-/dist
+/dist
+.envrc
+.env

CHANGELOG.md

@@ -8,6 +8,42 @@ Operators, you should copy/paste content of this content straight to your projec
 
 If you were at `firehose-core` version `1.0.0` and are bumping to `1.1.0`, you should copy the content between those 2 version to your own repository, replacing placeholder value `fire{chain}` with your chain's own binary.
 
+## v1.0.0
+
+This is a major release.
+
+### Operators
+
+> [!IMPORTANT]
+> When upgrading your stack to firehose-core v1.0.0, be sure to upgrade all components simultaneously because the block encapsulation format has changed.
+> Blocks that are merged using the new merger will not be readable by previous versions.
+
+### Added
+
+* New binary `firecore` which can run all firehose components (`reader`, `reader-stdin`, `merger`, `relayer`, `firehose`, `substreams-tier1|2`) in a chain-agnostic way. This is not mandatory (it can still be used as a library) but strongly suggested when possible.
+
+* Current Limitations on Ethereum:
+  - The firecore `firehose` app does not support transforms (filters, header-only --for graph-node compatibility--) so you will want to continue running this app from `fireeth`
+  - The firecore `substreams` apps do not support eth_calls so you will want to continue running them from `fireeth`
+  - The firecore `reader` does not support the block format output by the current geth firehose instrumentation, so you will want to continue running it from `fireeth`
+
+* New BlockPoller library to facilitate the implementation of rpc-poller-based chains, taking care of managing reorgs
+
+* Considering that firehose-core is chain-agnostic, it's not aware of the different of the different block types. To be able to use tools around block decoding/printing, 
+  there are two ways to provide the type definition:
+  1. the 'protoregistry' package contains well-known block type definitions (ethereum, near, solana, bitcoin...), you won't need to provide anything in those cases.
+  2. for other types, you can provide additional protobuf files using `--proto-path` flag
+
+### Changed
+
+* Merged blocks storage format has been changed. Current blocks will continue to be decoded, but new merged blocks will not be readable by previous software versions.
+* The code from the following repositories have been merged into this repo. They will soon be archived.
+  * github.com/streamingfast/node-manager
+  * github.com/streamingfast/merger
+  * github.com/streamingfast/relayer
+  * github.com/streamingfast/firehose
+  * github.com/streamingfast/index-builder
+
 ## v0.2.4
 
 * Fixed SF_TRACING feature (regression broke the ability to specify a tracing endpoint)

Dockerfile

@@ -0,0 +1,26 @@
+FROM golang:1.21-alpine as build
+WORKDIR /app
+
+COPY go.mod go.sum ./
+RUN go mod download
+
+COPY . ./
+
+RUN go build ./cmd/firecore
+
+####
+
+FROM alpine:edge
+
+
+RUN apk --no-cache add \
+        ca-certificates htop iotop sysstat \
+        strace lsof curl jq tzdata
+
+RUN mkdir -p /app/ && curl -Lo /app/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/v0.4.12/grpc_health_probe-linux-amd64 && chmod +x /app/grpc_health_probe
+
+WORKDIR /app
+
+COPY --from=build /app/firecore /app/firecore
+
+ENTRYPOINT [ "/app/firecore" ]

README.md

@@ -8,7 +8,7 @@ This repository contains all the boilerplate code that is required to maintain t
 
 Firehose maintenance cost comes from two sides. First, there is the chain integration that needs to be maintained. This is done within the chain's code directly by the chain's core developers. The second side of things is the maintenance of the Golang part of the Firehose stack.
 
-Each chain creates its own Firehose Golang repository named `firehose-<chain>`. [Firehose-acme repository](https://github.com/streamingfast/firehose-acme) acts as an example of this. Firehose is composed of multiple smaller components that can be run independently and each of them has a set of CLI flags and other configuration parameters.
+Each chain creates its own Firehose Golang repository named `firehose-<chain>`. [Firehose-acme repository](https://github.com/streamingfast/firehose-core/firehose-acme) acts as an example of this. Firehose is composed of multiple smaller components that can be run independently and each of them has a set of CLI flags and other configuration parameters.
 
 The initial "Acme" template we had contained a lot of boilerplate code to properly configure and run the Firehose Golang stack. This meant that if we needed to add a new feature that required a new flag or change a flag default value or any kind of improvements, chain integrators that were maintaining their `firehose-<chain>` repository were in the obligation of tracking changes made in `firehose-acme` and apply those back on their repository by hand.
 
@@ -33,4 +33,4 @@ When bumping `firehose-core` to a breaking version, details of such upgrade will
 
 ### Build & CI
 
-The build and CI files are maintained for now in https://github.com/streamingfast/firehose-acme directly and should be updated manually from time to time from there.
+The build and CI files are maintained for now in https://github.com/streamingfast/firehose-core/firehose-acme directly and should be updated manually from time to time from there.

blockpoller/cursor.go

@@ -0,0 +1,74 @@
+package blockpoller
+
+import (
+	"github.com/streamingfast/bstream"
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
+	"go.uber.org/zap"
+)
+
+type State string
+
+const (
+	ContinuousSegState State = "CONTINUOUS"
+	IncompleteSegState State = "INCOMPLETE"
+)
+
+func (s State) String() string {
+	return string(s)
+}
+
+type cursor struct {
+	currentBlk           bstream.BlockRef
+	currentIncompleteSeg *bstream.BasicBlockRef
+	state                State
+	logger               *zap.Logger
+}
+
+func (s *cursor) addBlk(blk *pbbstream.Block, blockSeen bool, parentSeen bool) {
+	blkRef := blk.AsRef()
+	logger := s.logger.With(
+		zap.Stringer("blk", blkRef),
+		zap.Stringer("parent_blk", blk.PreviousRef()),
+		zap.Bool("seen_blk", blockSeen),
+		zap.Bool("seen_parent", parentSeen),
+		zap.Stringer("previous_state", s.state),
+	)
+	if s.currentIncompleteSeg != nil {
+		logger = logger.With(zap.Stringer("current_incomplete_seg", *s.currentIncompleteSeg))
+	} else {
+		logger = logger.With(zap.String("current_incomplete_seg", "none"))
+
+	}
+
+	if s.state == IncompleteSegState && blockSeen && parentSeen {
+		// if we are checking an incomplete segement, and we get a block that is already in the forkdb
+		// and whose parent is also in the forkdb, then we are back on a continuous segment
+		s.state = ContinuousSegState
+	}
+	s.currentBlk = blkRef
+	logger.Debug("received block", zap.Stringer("current_state", s.state))
+}
+
+func (s *cursor) getBlkSegmentNum() bstream.BlockRef {
+	if s.state == IncompleteSegState {
+		if s.currentIncompleteSeg == nil {
+			panic("current incomplete segment is nil, when cursor is incomplete segment, this should never happen")
+		}
+		return *s.currentIncompleteSeg
+	}
+	return s.currentBlk
+}
+
+func (s *cursor) blkIsConnectedToLib() {
+	s.state = ContinuousSegState
+	s.currentIncompleteSeg = nil
+}
+
+func (s *cursor) blkIsNotConnectedToLib() {
+	if s.state != IncompleteSegState {
+		s.state = IncompleteSegState
+		// we don't want to point the current blk since that will change
+		v := bstream.NewBlockRef(s.currentBlk.ID(), s.currentBlk.Num())
+		s.currentIncompleteSeg = &v
+	}
+}

blockpoller/fetcher.go

@@ -0,0 +1,11 @@
+package blockpoller
+
+import (
+	"context"
+
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
+)
+
+type BlockFetcher interface {
+	Fetch(ctx context.Context, blkNum uint64) (*pbbstream.Block, error)
+}

blockpoller/handler.go

@@ -0,0 +1,57 @@
+package blockpoller
+
+import (
+	"encoding/base64"
+	"fmt"
+	"strings"
+	"sync"
+
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
+)
+
+type BlockHandler interface {
+	Init()
+	Handle(blk *pbbstream.Block) error
+}
+
+var _ BlockHandler = (*FireBlockHandler)(nil)
+
+type FireBlockHandler struct {
+	blockTypeURL string
+	init         sync.Once
+}
+
+func NewFireBlockHandler(blockTypeURL string) *FireBlockHandler {
+	return &FireBlockHandler{
+		blockTypeURL: clean(blockTypeURL),
+	}
+}
+
+func (f *FireBlockHandler) Init() {
+	fmt.Println("FIRE INIT 1.0", f.blockTypeURL)
+}
+
+func (f *FireBlockHandler) Handle(b *pbbstream.Block) error {
+	typeURL := clean(b.Payload.TypeUrl)
+	if typeURL != f.blockTypeURL {
+		return fmt.Errorf("block type url %q does not match expected type %q", typeURL, f.blockTypeURL)
+	}
+
+	blockLine := fmt.Sprintf(
+		"FIRE BLOCK %d %s %d %s %d %d %s",
+		b.Number,
+		b.Id,
+		b.ParentNum,
+		b.ParentId,
+		b.LibNum,
+		b.Timestamp.AsTime().UnixNano(),
+		base64.StdEncoding.EncodeToString(b.Payload.Value),
+	)
+
+	fmt.Println(blockLine)
+	return nil
+}
+
+func clean(in string) string {
+	return strings.Replace(in, "type.googleapis.com/", "", 1)
+}

blockpoller/handler_test.go

@@ -0,0 +1,24 @@
+package blockpoller
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFireBlockHandler_clean(t *testing.T) {
+	tests := []struct {
+		in     string
+		expect string
+	}{
+		{"type.googleapis.com/sf.bstream.v2.Block", "sf.bstream.v2.Block"},
+		{"sf.bstream.v2.Block", "sf.bstream.v2.Block"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.in, func(t *testing.T) {
+			assert.Equal(t, test.expect, clean(test.in))
+		})
+	}
+
+}