add CHANGELOG, remove 'forked'-related flags, fix indexes

CHANGELOG.md

@@ -1 +1,26 @@
 # Changelog
+
+# v0.6.0
+
+Exposing new GRPC endpoints for firehose: `sf.firehose.v2.Stream/Blocks` and `sf.firehose.v2.Fetch/Block`
+
+## Breaking changes
+* One-block-file naming is different: you need to remove existing one-block-files when transitioning
+
+## Changed naming and flags
+* Changed binary name: `firehose-cosmos` becomes `firecosmos`
+* `firehose-block-index-sizes` becomes `common-block-index-sizes`
+* `firehose-block-index-url` becomes `common-index-store-url`
+* `common-blocks-store-url` becomes `common-merged-blocks-store-url`
+* `common-blockstream-addr` becomes `common-live-blocks-addr`
+* `common-oneblock-store-url` becomes `common-one-block-store-url`
+
+## Removed flags
+* `firehose-real-time-tolerance`, `firehose-rpc-head-tracker-url`, `firehose-static-head-tracker`, `firehose-tracker-offset` are removed
+* `merger-state-file`, `merger-max-one-block-operations-batch-size`, `merger-next-exclusive-highest-block-limit`, `merger-one-block-deletion-threads`, `merger-writers-leeway` are removed
+* `ingestor-merge-threshold-block-age`, `reader-wait-upload-complete-on-shutdown`
+* `relayer-buffer-size`, `relayer-merger-addr`, `relayer-min-start-offset`, `relayer-source-request-burst`
+
+## Added flags
+* `merger-stop-block`, `merger-time-between-store-pruning`
+* `reader-oneblock-suffix`, `reader-readiness-max-latency`

README.md

@@ -60,7 +60,6 @@ Available Commands:
 Flags:
       --common-auth-plugin string            Auth plugin URI, see streamingfast/dauth repository (default "null://")
       --common-merged-blocks-store-url string       Store URL (with prefix) where to read/write (default "file://{fh-data-dir}/storage/merged-blocks")
-      --common-forked-blocks-store-url string       Store URL (with prefix) where to read/write (default "file://{fh-data-dir}/storage/forked-blocks")
       --common-live-blocks-addr string       GRPC endpoint to get real-time blocks (default "0.0.0.0:9010")
       --common-first-streamable-block uint   First streamable block number
       --common-metering-plugin string        Metering plugin URI, see streamingfast/dmetering repository (default "null://")

cmd/firecosmos/cli/app_reader.go

@@ -78,7 +78,7 @@ func init() {
 	factoryFunc := func(runtime *launcher.Runtime) (launcher.App, error) {
 		sfDataDir := runtime.AbsDataDir
 
-		_, oneBlockStoreURL, _, err := GetCommonStoresURLs(runtime.AbsDataDir)
+		_, oneBlockStoreURL, err := GetCommonStoresURLs(runtime.AbsDataDir)
 		workingDir := MustReplaceDataDir(sfDataDir, viper.GetString("reader-working-dir"))
 		gprcListenAdrr := viper.GetString("reader-grpc-listen-addr")
 		batchStartBlockNum := viper.GetUint64("reader-start-block-num")

cmd/firecosmos/cli/defaults.go

@@ -21,7 +21,9 @@ var (
 	// Blocks store
 	MergedBlocksStoreURL string = "file://{fh-data-dir}/storage/merged-blocks"
 	OneBlockStoreURL     string = "file://{fh-data-dir}/storage/one-blocks"
-	ForkedBlockStoreURL  string = "file://{fh-data-dir}/storage/forked-blocks"
+
+	IndexStoreURL   string = "file://{fh-data-dir}/storage/block-index"
+	IndexStoreSizes        = []int{100000, 10000, 1000}
 
 	// Protocol defaults
 	FirstStreamableBlock uint64 = 0
@@ -48,10 +50,12 @@ func initCommonFlags(flags *pflag.FlagSet) {
 	// Common stores configuration flags
 	flags.String("common-merged-blocks-store-url", MergedBlocksStoreURL, "Store URL (with prefix) where to read/write")
 	flags.String("common-one-block-store-url", OneBlockStoreURL, "Store URL (with prefix) to read/write one-block files")
-	flags.String("common-forked-blocks-store-url", ForkedBlockStoreURL, "Store URL (with prefix) to read/write forked one-block files")
 	flags.String("common-live-blocks-addr", RelayerServingAddr, "GRPC endpoint to get real-time blocks")
 	flags.Uint64("common-first-streamable-block", FirstStreamableBlock, "First streamable block number")
 
+	flags.String("common-index-store-url", IndexStoreURL, "[COMMON] Store URL (with prefix) to read/write index files.")
+	flags.IntSlice("common-block-index-sizes", []int{100000, 100000, 10000, 1000}, "index bundle sizes that that are considered valid when looking for block indexes")
+
 	// Authentication, metering and rate limiter plugins
 	flags.String("common-auth-plugin", "null://", "Auth plugin URI, see streamingfast/dauth repository")
 	flags.String("common-metering-plugin", "null://", "Metering plugin URI, see streamingfast/dmetering repository")

cmd/firecosmos/cli/firehose.go

@@ -54,7 +54,7 @@ func init() {
 		}
 		dmetering.SetDefaultMeter(metering)
 
-		mergedBlocksStoreURL, oneBlocksStoreURL, forkedBlocksStoreURL, err := GetCommonStoresURLs(runtime.AbsDataDir)
+		mergedBlocksStoreURL, oneBlocksStoreURL, err := GetCommonStoresURLs(runtime.AbsDataDir)
 		if err != nil {
 			return nil, err
 		}
@@ -79,7 +79,7 @@ func init() {
 			&firehoseApp.Config{
 				MergedBlocksStoreURL:    mergedBlocksStoreURL,
 				OneBlocksStoreURL:       oneBlocksStoreURL,
-				ForkedBlocksStoreURL:    forkedBlocksStoreURL,
+				ForkedBlocksStoreURL:    ".", // no forked blocks here
 				BlockStreamAddr:         blockstreamAddr,
 				GRPCListenAddr:          viper.GetString("firehose-grpc-listen-addr"),
 				GRPCShutdownGracePeriod: grcpShutdownGracePeriod,
@@ -95,7 +95,7 @@ func init() {
 	launcher.RegisterApp(zlog, &launcher.AppDef{
 		ID:            "firehose",
 		Title:         "Block Firehose",
-		Description:   "Provides on-demand filtered blocks, depends on common-merged-blocks-store-url, common-forked-blocks-store-url and common-live-blocks-addr",
+		Description:   "Provides on-demand filtered blocks, depends on common-merged-blocks-store-url and common-live-blocks-addr",
 		RegisterFlags: registerFlags,
 		FactoryFunc:   factoryFunc,
 	})

cmd/firecosmos/cli/merger.go

@@ -13,7 +13,6 @@ func init() {
 	flags := func(cmd *cobra.Command) error {
 		cmd.Flags().Duration("merger-time-between-store-lookups", 1*time.Second, "Delay between source store polling (should be higher for remote storage)")
 		cmd.Flags().Duration("merger-time-between-store-pruning", time.Minute, "Delay between source store pruning loops")
-		cmd.Flags().Uint64("merger-prune-forked-blocks-after", 50000, "Number of blocks that must pass before we delete old forks (one-block-files lingering)")
 		cmd.Flags().String("merger-grpc-listen-addr", MergerServingAddr, "Address to listen for incoming gRPC requests")
 		cmd.Flags().Uint64("merger-stop-block", 0, "if non-zero, merger will trigger shutdown when blocks have been merged up to this block")
 		return nil
@@ -25,17 +24,17 @@ func init() {
 
 	factoryFunc := func(runtime *launcher.Runtime) (launcher.App, error) {
 
-		mergedBlocksStoreURL, oneBlocksStoreURL, forkedBlocksStoreURL, err := GetCommonStoresURLs(runtime.AbsDataDir)
+		mergedBlocksStoreURL, oneBlocksStoreURL, err := GetCommonStoresURLs(runtime.AbsDataDir)
 		if err != nil {
 			return nil, err
 		}
 		return mergerApp.New(&mergerApp.Config{
 
 			StorageOneBlockFilesPath:     oneBlocksStoreURL,
 			StorageMergedBlocksFilesPath: mergedBlocksStoreURL,
-			StorageForkedBlocksFilesPath: forkedBlocksStoreURL,
+			StorageForkedBlocksFilesPath: ".",
 			GRPCListenAddr:               viper.GetString("merger-grpc-listen-addr"),
-			PruneForkedBlocksAfter:       viper.GetUint64("merger-prune-forked-blocks-after"),
+			PruneForkedBlocksAfter:       99999999999, // no forked blocks here
 			StopBlock:                    viper.GetUint64("merger-stop-block"),
 			TimeBetweenPruning:           viper.GetDuration("merger-time-between-store-pruning"),
 			TimeBetweenPolling:           viper.GetDuration("merger-time-between-store-lookups"),

cmd/firecosmos/cli/relayer.go

@@ -23,7 +23,7 @@ func init() {
 		if len(sources) == 0 {
 			return nil, errors.New("relayer sources are empty")
 		}
-		_, oneBlocksStoreURL, _, err := GetCommonStoresURLs(runtime.AbsDataDir)
+		_, oneBlocksStoreURL, err := GetCommonStoresURLs(runtime.AbsDataDir)
 		if err != nil {
 			return nil, err
 		}

cmd/firecosmos/cli/utils.go

@@ -18,18 +18,14 @@ import (
 var commonStoresCreated bool
 var indexStoreCreated bool
 
-func GetCommonStoresURLs(dataDir string) (mergedBlocksStoreURL, oneBlocksStoreURL, forkedBlocksStoreURL string, err error) {
+func GetCommonStoresURLs(dataDir string) (mergedBlocksStoreURL, oneBlocksStoreURL string, err error) {
 	mergedBlocksStoreURL = MustReplaceDataDir(dataDir, viper.GetString("common-merged-blocks-store-url"))
 	oneBlocksStoreURL = MustReplaceDataDir(dataDir, viper.GetString("common-one-block-store-url"))
-	forkedBlocksStoreURL = MustReplaceDataDir(dataDir, viper.GetString("common-forked-blocks-store-url"))
 
 	if commonStoresCreated {
 		return
 	}
 
-	if err = mkdirStorePathIfLocal(forkedBlocksStoreURL); err != nil {
-		return
-	}
 	if err = mkdirStorePathIfLocal(oneBlocksStoreURL); err != nil {
 		return
 	}