fix: state listener observe writes at wrong time (#13516)

* fix: state listener observe writes at wrong time

Closes: #13457

Currently state listener is notified when the cache store write, which happens in commit event only, which breaks the current design.
The solution (as discussed in the issue) is to listen state writes on rootmulti store only.

It also changes the file streamer to output single data file for the writes in the whole block, since we can't distinguish writes from different stage of abci events.

It adds new config items for file streamer:
- streamers.file.output-metadata
- streamers.file.stop-node-on-error
- streamers.file.fsync

* synchronous abci call, and format doc

* fix comment

* update file streamer readme and fix typos

* typo

* fix: state listener observe writes at wrong time

Closes: #13457

Currently state listener is notified when the cache store write, which happens in commit event only, which breaks the current design.
The solution (as discussed in the issue) is to listen state writes on rootmulti store only.

It also changes the file streamer to output single data file for the writes in the whole block, since we can't distinguish writes from different stage of abci events.

It adds new config items for file streamer:
- streamers.file.output-metadata
- streamers.file.stop-node-on-error
- streamers.file.fsync

synchronous abci call, and format doc

fix comment

update file streamer readme and fix typos

typo

* improve UX of file streamer, make it immediately usable after enabled

- set default value to write_dir.
- make write_dir based on home directory by default.
- auto-create the directory if not exists.

* get homePage from opts

Co-authored-by: Marko <marbar3778@yahoo.com>

store/streaming/constructor.go

@@ -2,21 +2,25 @@ package streaming
 
 import (
 	"fmt"
+	"os"
+	"path"
 	"strings"
 	"sync"
 
 	"github.com/cosmos/cosmos-sdk/baseapp"
+	"github.com/cosmos/cosmos-sdk/client/flags"
 	"github.com/cosmos/cosmos-sdk/codec"
 	serverTypes "github.com/cosmos/cosmos-sdk/server/types"
 	"github.com/cosmos/cosmos-sdk/store/streaming/file"
 	"github.com/cosmos/cosmos-sdk/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/tendermint/tendermint/libs/log"
 
 	"github.com/spf13/cast"
 )
 
 // ServiceConstructor is used to construct a streaming service
-type ServiceConstructor func(serverTypes.AppOptions, []types.StoreKey, codec.BinaryCodec) (baseapp.StreamingService, error)
+type ServiceConstructor func(serverTypes.AppOptions, []types.StoreKey, codec.BinaryCodec, log.Logger) (baseapp.StreamingService, error)
 
 // ServiceType enum for specifying the type of StreamingService
 type ServiceType int
@@ -28,9 +32,13 @@ const (
 
 // Streaming option keys
 const (
-	OptStreamersFilePrefix   = "streamers.file.prefix"
-	OptStreamersFileWriteDir = "streamers.file.write_dir"
-	OptStoreStreamers        = "store.streamers"
+	OptStreamersFilePrefix          = "streamers.file.prefix"
+	OptStreamersFileWriteDir        = "streamers.file.write_dir"
+	OptStreamersFileOutputMetadata  = "streamers.file.output-metadata"
+	OptStreamersFileStopNodeOnError = "streamers.file.stop-node-on-error"
+	OptStreamersFileFsync           = "streamers.file.fsync"
+
+	OptStoreStreamers = "store.streamers"
 )
 
 // ServiceTypeFromString returns the streaming.ServiceType corresponding to the
@@ -83,11 +91,28 @@ func NewFileStreamingService(
 	opts serverTypes.AppOptions,
 	keys []types.StoreKey,
 	marshaller codec.BinaryCodec,
+	logger log.Logger,
 ) (baseapp.StreamingService, error) {
+	homePath := cast.ToString(opts.Get(flags.FlagHome))
 	filePrefix := cast.ToString(opts.Get(OptStreamersFilePrefix))
 	fileDir := cast.ToString(opts.Get(OptStreamersFileWriteDir))
+	outputMetadata := cast.ToBool(opts.Get(OptStreamersFileOutputMetadata))
+	stopNodeOnErr := cast.ToBool(opts.Get(OptStreamersFileStopNodeOnError))
+	fsync := cast.ToBool(opts.Get(OptStreamersFileFsync))
 
-	return file.NewStreamingService(fileDir, filePrefix, keys, marshaller)
+	// relative path is based on node home directory.
+	if !path.IsAbs(fileDir) {
+		fileDir = path.Join(homePath, fileDir)
+	}
+
+	// try to create output directory if not exists.
+	if _, err := os.Stat(fileDir); os.IsNotExist(err) {
+		if err = os.MkdirAll(fileDir, os.ModePerm); err != nil {
+			return nil, err
+		}
+	}
+
+	return file.NewStreamingService(fileDir, filePrefix, keys, marshaller, logger, outputMetadata, stopNodeOnErr, fsync)
 }
 
 // LoadStreamingServices is a function for loading StreamingServices onto the
@@ -98,6 +123,7 @@ func LoadStreamingServices(
 	bApp *baseapp.BaseApp,
 	appOpts serverTypes.AppOptions,
 	appCodec codec.BinaryCodec,
+	logger log.Logger,
 	keys map[string]*types.KVStoreKey,
 ) ([]baseapp.StreamingService, *sync.WaitGroup, error) {
 	// waitgroup and quit channel for optional shutdown coordination of the streaming service(s)
@@ -145,7 +171,7 @@ func LoadStreamingServices(
 
 		// Generate the streaming service using the constructor, appOptions, and the
 		// StoreKeys we want to expose.
-		streamingService, err := constructor(appOpts, exposeStoreKeys, appCodec)
+		streamingService, err := constructor(appOpts, exposeStoreKeys, appCodec, logger)
 		if err != nil {
 			// Close any services we may have already spun up before hitting the error
 			// on this one.

store/streaming/constructor_test.go

@@ -22,7 +22,13 @@ import (
 
 type fakeOptions struct{}
 
-func (f *fakeOptions) Get(string) interface{} { return nil }
+func (f *fakeOptions) Get(key string) interface{} {
+	if key == "streamers.file.write_dir" {
+		return "data/file_streamer"
+
+	}
+	return nil
+}
 
 var (
 	mockOptions       = new(fakeOptions)
@@ -40,7 +46,7 @@ func TestStreamingServiceConstructor(t *testing.T) {
 	var expectedType streaming.ServiceConstructor
 	require.IsType(t, expectedType, constructor)
 
-	serv, err := constructor(mockOptions, mockKeys, testMarshaller)
+	serv, err := constructor(mockOptions, mockKeys, testMarshaller, log.NewNopLogger())
 	require.Nil(t, err)
 	require.IsType(t, &file.StreamingService{}, serv)
 	listeners := serv.Listeners()
@@ -78,7 +84,7 @@ func TestLoadStreamingServices(t *testing.T) {
 
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
-			activeStreamers, _, err := streaming.LoadStreamingServices(bApp, tc.appOpts, encCdc.Codec, keys)
+			activeStreamers, _, err := streaming.LoadStreamingServices(bApp, tc.appOpts, encCdc.Codec, log.NewNopLogger(), keys)
 			require.NoError(t, err)
 			require.Equal(t, tc.activeStreamersLen, len(activeStreamers))
 		})
@@ -95,6 +101,8 @@ func (ao streamingAppOptions) Get(o string) interface{} {
 		return []string{"file"}
 	case "streamers.file.keys":
 		return ao.keys
+	case "streamers.file.write_dir":
+		return "data/file_streamer"
 	default:
 		return nil
 	}

store/streaming/file/README.md

@@ -25,42 +25,67 @@ We turn the service on by adding its name, "file", to `store.streamers`- the lis
 
 In `streamers.file` we include three configuration parameters for the file streaming service:
 
-1. `streamers.x.keys` contains the list of `StoreKey` names for the KVStores to expose using this service.
-In order to expose *all* KVStores, we can include `*` in this list. An empty list is equivalent to turning the service off.
+1. `streamers.file.keys` contains the list of `StoreKey` names for the KVStores to expose using this service.
+    In order to expose *all* KVStores, we can include `*` in this list. An empty list is equivalent to turning the service off.
 2. `streamers.file.write_dir` contains the path to the directory to write the files to.
 3. `streamers.file.prefix` contains an optional prefix to prepend to the output files to prevent potential collisions
-with other App `StreamingService` output files.
+    with other App `StreamingService` output files.
+4. `streamers.file.output-metadata` specifies if output the metadata file, otherwise only data file is outputted.
+5. `streamers.file.stop-node-on-error` specifies if propagate the error to consensus state machine, it's nesserary for data integrity when node restarts.
+6. `streamers.file.fsync` specifies if call fsync after writing the files, it's nesserary for data integrity when system crash, but slows down the commit time.
 
 ### Encoding
 
-For each pair of `BeginBlock` requests and responses, a file is created and named `block-{N}-begin`, where N is the block number.
-At the head of this file the length-prefixed protobuf encoded `BeginBlock` request is written.
-At the tail of this file the length-prefixed protobuf encoded `BeginBlock` response is written.
-In between these two encoded messages, the state changes that occurred due to the `BeginBlock` request are written chronologically as
-a series of length-prefixed protobuf encoded `StoreKVPair`s representing `Set` and `Delete` operations within the KVStores the service
-is configured to listen to.
+For each block, two files are created and names `block-{N}-meta` and `block-{N}-data`, where `N` is the block number.
 
-For each pair of `DeliverTx` requests and responses, a file is created and named `block-{N}-tx-{M}` where N is the block number and M
-is the tx number in the block (i.e. 0, 1, 2...).
-At the head of this file the length-prefixed protobuf encoded `DeliverTx` request is written.
-At the tail of this file the length-prefixed protobuf encoded `DeliverTx` response is written.
-In between these two encoded messages, the state changes that occurred due to the `DeliverTx` request are written chronologically as
-a series of length-prefixed protobuf encoded `StoreKVPair`s representing `Set` and `Delete` operations within the KVStores the service
-is configured to listen to.
+The meta file contains the protobuf encoded message `BlockMetadata` which contains the abci event requests and responses of the block:
 
-For each pair of `EndBlock` requests and responses, a file is created and named `block-{N}-end`, where N is the block number.
-At the head of this file the length-prefixed protobuf encoded `EndBlock` request is written.
-At the tail of this file the length-prefixed protobuf encoded `EndBlock` response is written.
-In between these two encoded messages, the state changes that occurred due to the `EndBlock` request are written chronologically as
-a series of length-prefixed protobuf encoded `StoreKVPair`s representing `Set` and `Delete` operations within the KVStores the service
-is configured to listen to.
+```protobuf
+message BlockMetadata {
+    message DeliverTx {
+        tendermint.abci.RequestDeliverTx request = 1;
+        tendermint.abci.ResponseDeliverTx response = 2;
+    }
+    tendermint.abci.RequestBeginBlock request_begin_block = 1;
+    tendermint.abci.ResponseBeginBlock response_begin_block = 2;
+    repeated DeliverTx deliver_txs = 3;
+    tendermint.abci.RequestEndBlock request_end_block = 4;
+    tendermint.abci.ResponseEndBlock response_end_block = 5;
+    tendermint.abci.ResponseCommit response_commit = 6;
+}
+```
+
+The data file contains a series of length-prefixed protobuf encoded `StoreKVPair`s representing `Set` and `Delete` operations within the KVStores during the execution of block.
+
+Both meta and data files are prefixed with the length of the data content for consumer to detect completeness of the file, the length is encoded as 8 bytes with big endianness.
+
+The files are written at abci commit event, by default the error happens will be propagated to interuppted consensus state machine, but fsync is not called, it'll have good performance but have the risk of lossing data in face of rare event of system crash.
 
 ### Decoding
 
-To decode the files written in the above format we read all the bytes from a given file into memory and segment them into proto
-messages based on the length-prefixing of each message. Once segmented, it is known that the first message is the ABCI request,
-the last message is the ABCI response, and that every message in between is a `StoreKVPair`. This enables us to decode each segment into
-the appropriate message type.
+The pseudo-code for decoding is like this:
 
-The type of ABCI req/res, the block height, and the transaction index (where relevant) is known
-from the file name, and the KVStore each `StoreKVPair` originates from is known since the `StoreKey` is included as a field in the proto message.
+```python
+def decode_meta_file(file):
+  bz = file.read(8)
+  if len(bz) < 8:
+    raise "incomplete file exception"
+  size = int.from_bytes(bz, 'big')
+
+  if file.size != size + 8:
+    raise "incomplete file exception"
+
+  return decode_protobuf_message(BlockMetadata, file)
+
+def decode_data_file(file):
+  bz = file.read(8)
+  if len(bz) < 8:
+    raise "incomplete file exception"
+  size = int.from_bytes(bz, 'big')
+
+  if file.size != size + 8:
+    raise "incomplete file exception"
+
+  while not file.eof():
+    yield decode_length_prefixed_protobuf_message(StoreKVStore, file)
+```

store/streaming/file/service.go

@@ -1,65 +1,57 @@
 package file
 
 import (
+	"bytes"
 	"context"
-	"errors"
 	"fmt"
+	"io"
 	"os"
 	"path"
-	"path/filepath"
+	"sort"
 	"sync"
 
 	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/libs/log"
 
 	"github.com/cosmos/cosmos-sdk/baseapp"
 	"github.com/cosmos/cosmos-sdk/codec"
 	"github.com/cosmos/cosmos-sdk/store/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 )
 
 var _ baseapp.StreamingService = &StreamingService{}
 
 // StreamingService is a concrete implementation of StreamingService that writes state changes out to files
 type StreamingService struct {
-	listeners          map[types.StoreKey][]types.WriteListener // the listeners that will be initialized with BaseApp
-	srcChan            <-chan []byte                            // the channel that all the WriteListeners write their data out to
-	filePrefix         string                                   // optional prefix for each of the generated files
-	writeDir           string                                   // directory to write files into
-	codec              codec.BinaryCodec                        // marshaller used for re-marshalling the ABCI messages to write them out to the destination files
-	stateCache         [][]byte                                 // cache the protobuf binary encoded StoreKVPairs in the order they are received
-	stateCacheLock     *sync.Mutex                              // mutex for the state cache
-	currentBlockNumber int64                                    // the current block number
-	currentTxIndex     int64                                    // the index of the current tx
-	quitChan           chan struct{}                            // channel to synchronize closure
-}
+	storeListeners []*types.MemoryListener // a series of KVStore listeners for each KVStore
+	filePrefix     string                  // optional prefix for each of the generated files
+	writeDir       string                  // directory to write files into
+	codec          codec.BinaryCodec       // marshaller used for re-marshalling the ABCI messages to write them out to the destination files
+	logger         log.Logger
 
-// IntermediateWriter is used so that we do not need to update the underlying io.Writer
-// inside the StoreKVPairWriteListener everytime we begin writing to a new file
-type IntermediateWriter struct {
-	outChan chan<- []byte
-}
-
-// NewIntermediateWriter create an instance of an IntermediateWriter that sends to the provided channel
-func NewIntermediateWriter(outChan chan<- []byte) *IntermediateWriter {
-	return &IntermediateWriter{
-		outChan: outChan,
-	}
-}
-
-// Write satisfies io.Writer
-func (iw *IntermediateWriter) Write(b []byte) (int, error) {
-	iw.outChan <- b
-	return len(b), nil
+	currentBlockNumber int64
+	blockMetadata      types.BlockMetadata
+	// if write the metadata file, otherwise only data file is outputted.
+	outputMetadata bool
+	// if true, when commit failed it will panic and stop the consensus state machine to ensure the
+	// eventual consistency of the output, otherwise the error is ignored and have the risk of lossing data.
+	stopNodeOnErr bool
+	// if true, the file.Sync() is called to make sure the data is persisted onto disk, otherwise it risks lossing data when system crash.
+	fsync bool
 }
 
 // NewStreamingService creates a new StreamingService for the provided writeDir, (optional) filePrefix, and storeKeys
-func NewStreamingService(writeDir, filePrefix string, storeKeys []types.StoreKey, c codec.BinaryCodec) (*StreamingService, error) {
-	listenChan := make(chan []byte)
-	iw := NewIntermediateWriter(listenChan)
-	listener := types.NewStoreKVPairWriteListener(iw, c)
-	listeners := make(map[types.StoreKey][]types.WriteListener, len(storeKeys))
+func NewStreamingService(writeDir, filePrefix string, storeKeys []types.StoreKey, c codec.BinaryCodec, logger log.Logger, outputMetadata bool, stopNodeOnErr bool, fsync bool) (*StreamingService, error) {
+	// sort storeKeys for deterministic output
+	sort.SliceStable(storeKeys, func(i, j int) bool {
+		return storeKeys[i].Name() < storeKeys[j].Name()
+	})
+
+	listeners := make([]*types.MemoryListener, len(storeKeys))
 	// in this case, we are using the same listener for each Store
-	for _, key := range storeKeys {
-		listeners[key] = append(listeners[key], listener)
+	for i, key := range storeKeys {
+		listeners[i] = types.NewMemoryListener(key)
 	}
 	// check that the writeDir exists and is writable so that we can catch the error here at initialization if it is not
 	// we don't open a dstFile until we receive our first ABCI message
@@ -67,13 +59,14 @@ func NewStreamingService(writeDir, filePrefix string, storeKeys []types.StoreKey
 		return nil, err
 	}
 	return &StreamingService{
-		listeners:      listeners,
-		srcChan:        listenChan,
+		storeListeners: listeners,
 		filePrefix:     filePrefix,
 		writeDir:       writeDir,
 		codec:          c,
-		stateCache:     make([][]byte, 0),
-		stateCacheLock: new(sync.Mutex),
+		logger:         logger,
+		outputMetadata: outputMetadata,
+		stopNodeOnErr:  stopNodeOnErr,
+		fsync:          fsync,
 	}, nil
 }
 
@@ -81,201 +74,106 @@ func NewStreamingService(writeDir, filePrefix string, storeKeys []types.StoreKey
 // It returns the StreamingService's underlying WriteListeners
 // Use for registering the underlying WriteListeners with the BaseApp
 func (fss *StreamingService) Listeners() map[types.StoreKey][]types.WriteListener {
-	return fss.listeners
+	listeners := make(map[types.StoreKey][]types.WriteListener, len(fss.storeListeners))
+	for _, listener := range fss.storeListeners {
+		listeners[listener.StoreKey()] = []types.WriteListener{listener}
+	}
+	return listeners
 }
 
 // ListenBeginBlock satisfies the baseapp.ABCIListener interface
 // It writes the received BeginBlock request and response and the resulting state changes
 // out to a file as described in the above the naming schema
 func (fss *StreamingService) ListenBeginBlock(ctx context.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) (rerr error) {
-	// generate the new file
-	dstFile, err := fss.openBeginBlockFile(req)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		cerr := dstFile.Close()
-		if rerr == nil {
-			rerr = cerr
-		}
-	}()
-
-	// write req to file
-	lengthPrefixedReqBytes, err := fss.codec.MarshalLengthPrefixed(&req)
-	if err != nil {
-		return err
-	}
-	if _, err = dstFile.Write(lengthPrefixedReqBytes); err != nil {
-		return err
-	}
-	// write all state changes cached for this stage to file
-	fss.stateCacheLock.Lock()
-	for _, stateChange := range fss.stateCache {
-		if _, err = dstFile.Write(stateChange); err != nil {
-			fss.stateCache = nil
-			fss.stateCacheLock.Unlock()
-			return err
-		}
-	}
-	// reset cache
-	fss.stateCache = nil
-	fss.stateCacheLock.Unlock()
-	// write res to file
-	lengthPrefixedResBytes, err := fss.codec.MarshalLengthPrefixed(&res)
-	if err != nil {
-		return err
-	}
-	_, err = dstFile.Write(lengthPrefixedResBytes)
-	return err
-}
-
-func (fss *StreamingService) openBeginBlockFile(req abci.RequestBeginBlock) (*os.File, error) {
-	fss.currentBlockNumber = req.GetHeader().Height
-	fss.currentTxIndex = 0
-	fileName := fmt.Sprintf("block-%d-begin", fss.currentBlockNumber)
-	if fss.filePrefix != "" {
-		fileName = fmt.Sprintf("%s-%s", fss.filePrefix, fileName)
-	}
-	return os.OpenFile(filepath.Join(fss.writeDir, fileName), os.O_CREATE|os.O_WRONLY, 0o600)
+	fss.blockMetadata.RequestBeginBlock = &req
+	fss.blockMetadata.ResponseBeginBlock = &res
+	fss.currentBlockNumber = req.Header.Height
+	return nil
 }
 
 // ListenDeliverTx satisfies the baseapp.ABCIListener interface
 // It writes the received DeliverTx request and response and the resulting state changes
 // out to a file as described in the above the naming schema
 func (fss *StreamingService) ListenDeliverTx(ctx context.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) (rerr error) {
-	// generate the new file
-	dstFile, err := fss.openDeliverTxFile()
-	if err != nil {
-		return err
-	}
-	defer func() {
-		cerr := dstFile.Close()
-		if rerr == nil {
-			rerr = cerr
-		}
-	}()
-
-	// write req to file
-	lengthPrefixedReqBytes, err := fss.codec.MarshalLengthPrefixed(&req)
-	if err != nil {
-		return err
-	}
-	if _, err = dstFile.Write(lengthPrefixedReqBytes); err != nil {
-		return err
-	}
-	// write all state changes cached for this stage to file
-	fss.stateCacheLock.Lock()
-	for _, stateChange := range fss.stateCache {
-		if _, err = dstFile.Write(stateChange); err != nil {
-			fss.stateCache = nil
-			fss.stateCacheLock.Unlock()
-			return err
-		}
-	}
-	// reset cache
-	fss.stateCache = nil
-	fss.stateCacheLock.Unlock()
-	// write res to file
-	lengthPrefixedResBytes, err := fss.codec.MarshalLengthPrefixed(&res)
-	if err != nil {
-		return err
-	}
-	_, err = dstFile.Write(lengthPrefixedResBytes)
-	return err
-}
-
-func (fss *StreamingService) openDeliverTxFile() (*os.File, error) {
-	fileName := fmt.Sprintf("block-%d-tx-%d", fss.currentBlockNumber, fss.currentTxIndex)
-	if fss.filePrefix != "" {
-		fileName = fmt.Sprintf("%s-%s", fss.filePrefix, fileName)
-	}
-	fss.currentTxIndex++
-	return os.OpenFile(filepath.Join(fss.writeDir, fileName), os.O_CREATE|os.O_WRONLY, 0o600)
+	fss.blockMetadata.DeliverTxs = append(fss.blockMetadata.DeliverTxs, &types.BlockMetadata_DeliverTx{
+		Request:  &req,
+		Response: &res,
+	})
+	return nil
 }
 
 // ListenEndBlock satisfies the baseapp.ABCIListener interface
 // It writes the received EndBlock request and response and the resulting state changes
 // out to a file as described in the above the naming schema
 func (fss *StreamingService) ListenEndBlock(ctx context.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) (rerr error) {
-	// generate the new file
-	dstFile, err := fss.openEndBlockFile()
-	if err != nil {
-		return err
-	}
-	defer func() {
-		cerr := dstFile.Close()
-		if rerr == nil {
-			rerr = cerr
-		}
-	}()
+	fss.blockMetadata.RequestEndBlock = &req
+	fss.blockMetadata.ResponseEndBlock = &res
+	return nil
+}
 
-	// write req to file
-	lengthPrefixedReqBytes, err := fss.codec.MarshalLengthPrefixed(&req)
+// ListenEndBlock satisfies the baseapp.ABCIListener interface
+func (fss *StreamingService) ListenCommit(ctx context.Context, res abci.ResponseCommit) error {
+	err := fss.doListenCommit(ctx, res)
 	if err != nil {
-		return err
-	}
-	if _, err = dstFile.Write(lengthPrefixedReqBytes); err != nil {
-		return err
-	}
-	// write all state changes cached for this stage to file
-	fss.stateCacheLock.Lock()
-	for _, stateChange := range fss.stateCache {
-		if _, err = dstFile.Write(stateChange); err != nil {
-			fss.stateCache = nil
-			fss.stateCacheLock.Unlock()
+		fss.logger.Error("Commit listening hook failed", "height", fss.currentBlockNumber, "err", err)
+		if fss.stopNodeOnErr {
 			return err
 		}
 	}
-	// reset cache
-	fss.stateCache = nil
-	fss.stateCacheLock.Unlock()
-	// write res to file
-	lengthPrefixedResBytes, err := fss.codec.MarshalLengthPrefixed(&res)
-	if err != nil {
-		return err
-	}
-	_, err = dstFile.Write(lengthPrefixedResBytes)
-	return err
+	return nil
 }
 
-func (fss *StreamingService) openEndBlockFile() (*os.File, error) {
-	fileName := fmt.Sprintf("block-%d-end", fss.currentBlockNumber)
+func (fss *StreamingService) doListenCommit(ctx context.Context, res abci.ResponseCommit) (err error) {
+	fss.blockMetadata.ResponseCommit = &res
+
+	// write to target files, the file size is written at the beginning, which can be used to detect completeness.
+	metaFileName := fmt.Sprintf("block-%d-meta", fss.currentBlockNumber)
+	dataFileName := fmt.Sprintf("block-%d-data", fss.currentBlockNumber)
 	if fss.filePrefix != "" {
-		fileName = fmt.Sprintf("%s-%s", fss.filePrefix, fileName)
+		metaFileName = fmt.Sprintf("%s-%s", fss.filePrefix, metaFileName)
+		dataFileName = fmt.Sprintf("%s-%s", fss.filePrefix, dataFileName)
 	}
-	return os.OpenFile(filepath.Join(fss.writeDir, fileName), os.O_CREATE|os.O_WRONLY, 0o600)
+
+	if fss.outputMetadata {
+		bz, err := fss.codec.Marshal(&fss.blockMetadata)
+		if err != nil {
+			return err
+		}
+		if err := writeLengthPrefixedFile(path.Join(fss.writeDir, metaFileName), bz, fss.fsync); err != nil {
+			return err
+		}
+	}
+
+	var buf bytes.Buffer
+	if err := fss.writeBlockData(&buf); err != nil {
+		return err
+	}
+	return writeLengthPrefixedFile(path.Join(fss.writeDir, dataFileName), buf.Bytes(), fss.fsync)
+}
+
+func (fss *StreamingService) writeBlockData(writer io.Writer) error {
+	for _, listener := range fss.storeListeners {
+		cache := listener.PopStateCache()
+		for i := range cache {
+			bz, err := fss.codec.MarshalLengthPrefixed(&cache[i])
+			if err != nil {
+				return err
+			}
+			if _, err = writer.Write(bz); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
 }
 
 // Stream satisfies the baseapp.StreamingService interface
-// It spins up a goroutine select loop which awaits length-prefixed binary encoded KV pairs
-// and caches them in the order they were received
-// returns an error if it is called twice
 func (fss *StreamingService) Stream(wg *sync.WaitGroup) error {
-	if fss.quitChan != nil {
-		return errors.New("`Stream` has already been called. The stream needs to be closed before it can be started again")
-	}
-	fss.quitChan = make(chan struct{})
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		for {
-			select {
-			case <-fss.quitChan:
-				fss.quitChan = nil
-				return
-			case by := <-fss.srcChan:
-				fss.stateCacheLock.Lock()
-				fss.stateCache = append(fss.stateCache, by)
-				fss.stateCacheLock.Unlock()
-			}
-		}
-	}()
 	return nil
 }
 
 // Close satisfies the io.Closer interface, which satisfies the baseapp.StreamingService interface
 func (fss *StreamingService) Close() error {
-	close(fss.quitChan)
 	return nil
 }
 
@@ -288,3 +186,32 @@ func isDirWriteable(dir string) error {
 	}
 	return os.Remove(f)
 }
+
+func writeLengthPrefixedFile(path string, data []byte, fsync bool) (err error) {
+	var f *os.File
+	f, err = os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
+	if err != nil {
+		return sdkerrors.Wrapf(err, "open file failed: %s", path)
+	}
+	defer func() {
+		// avoid overriding the real error with file close error
+		if err1 := f.Close(); err1 != nil && err == nil {
+			err = sdkerrors.Wrapf(err, "close file failed: %s", path)
+		}
+	}()
+	_, err = f.Write(sdk.Uint64ToBigEndian(uint64(len(data))))
+	if err != nil {
+		return sdkerrors.Wrapf(err, "write length prefix failed: %s", path)
+	}
+	_, err = f.Write(data)
+	if err != nil {
+		return sdkerrors.Wrapf(err, "write block data failed: %s", path)
+	}
+	if fsync {
+		err = f.Sync()
+		if err != nil {
+			return sdkerrors.Wrapf(err, "fsync failed: %s", path)
+		}
+	}
+	return
+}

store/streaming/file/service_test.go

@@ -2,6 +2,7 @@ package file
 
 import (
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -10,6 +11,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/libs/log"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 
 	"github.com/cosmos/cosmos-sdk/codec"
@@ -56,6 +58,10 @@ var (
 		ConsensusParamUpdates: &tmproto.ConsensusParams{},
 		ValidatorUpdates:      []abci.ValidatorUpdate{},
 	}
+	testCommitRes = abci.ResponseCommit{
+		Data:         []byte{1},
+		RetainHeight: 0,
+	}
 	mockTxBytes1      = []byte{9, 8, 7, 6, 5, 4, 3, 2, 1}
 	testDeliverTxReq1 = abci.RequestDeliverTx{
 		Tx: mockTxBytes1,
@@ -104,25 +110,6 @@ var (
 	mockValue3 = []byte{5, 4, 3}
 )
 
-func TestIntermediateWriter(t *testing.T) {
-	outChan := make(chan []byte, 0)
-	iw := NewIntermediateWriter(outChan)
-	require.IsType(t, &IntermediateWriter{}, iw)
-	testBytes := []byte{1, 2, 3, 4, 5}
-	var length int
-	var err error
-	waitChan := make(chan struct{}, 0)
-	go func() {
-		length, err = iw.Write(testBytes)
-		waitChan <- struct{}{}
-	}()
-	receivedBytes := <-outChan
-	<-waitChan
-	require.Equal(t, len(testBytes), length)
-	require.Equal(t, testBytes, receivedBytes)
-	require.Nil(t, err)
-}
-
 func TestFileStreamingService(t *testing.T) {
 	if os.Getenv("CI") != "" {
 		t.Skip("Skipping TestFileStreamingService in CI environment")
@@ -132,29 +119,23 @@ func TestFileStreamingService(t *testing.T) {
 	defer os.RemoveAll(testDir)
 
 	testKeys := []types.StoreKey{mockStoreKey1, mockStoreKey2}
-	testStreamingService, err = NewStreamingService(testDir, testPrefix, testKeys, testMarshaller)
+	testStreamingService, err = NewStreamingService(testDir, testPrefix, testKeys, testMarshaller, log.NewNopLogger(), true, false, false)
 	require.Nil(t, err)
 	require.IsType(t, &StreamingService{}, testStreamingService)
 	require.Equal(t, testPrefix, testStreamingService.filePrefix)
 	require.Equal(t, testDir, testStreamingService.writeDir)
 	require.Equal(t, testMarshaller, testStreamingService.codec)
-	testListener1 = testStreamingService.listeners[mockStoreKey1][0]
-	testListener2 = testStreamingService.listeners[mockStoreKey2][0]
+	testListener1 = testStreamingService.storeListeners[0]
+	testListener2 = testStreamingService.storeListeners[1]
 	wg := new(sync.WaitGroup)
 	testStreamingService.Stream(wg)
-	testListenBeginBlock(t)
-	testListenDeliverTx1(t)
-	testListenDeliverTx2(t)
-	testListenEndBlock(t)
+	testListenBlock(t)
 	testStreamingService.Close()
 	wg.Wait()
 }
 
-func testListenBeginBlock(t *testing.T) {
-	expectedBeginBlockReqBytes, err := testMarshaller.Marshal(&testBeginBlockReq)
-	require.Nil(t, err)
-	expectedBeginBlockResBytes, err := testMarshaller.Marshal(&testBeginBlockRes)
-	require.Nil(t, err)
+func testListenBlock(t *testing.T) {
+	var expectKVPairsStore1, expectKVPairsStore2 [][]byte
 
 	// write state changes
 	testListener1.OnWrite(mockStoreKey1, mockKey1, mockValue1, false)
@@ -183,162 +164,102 @@ func testListenBeginBlock(t *testing.T) {
 		Delete:   false,
 	})
 	require.Nil(t, err)
+	expectKVPairsStore1 = append(expectKVPairsStore1, expectedKVPair1, expectedKVPair3)
+	expectKVPairsStore2 = append(expectKVPairsStore2, expectedKVPair2)
 
 	// send the ABCI messages
 	err = testStreamingService.ListenBeginBlock(emptyContext, testBeginBlockReq, testBeginBlockRes)
 	require.Nil(t, err)
 
-	// load the file, checking that it was created with the expected name
-	fileName := fmt.Sprintf("%s-block-%d-begin", testPrefix, testBeginBlockReq.GetHeader().Height)
-	fileBytes, err := readInFile(fileName)
-	require.Nil(t, err)
-
-	// segment the file into the separate gRPC messages and check the correctness of each
-	segments, err := segmentBytes(fileBytes)
-	require.Nil(t, err)
-	require.Equal(t, 5, len(segments))
-	require.Equal(t, expectedBeginBlockReqBytes, segments[0])
-	require.Equal(t, expectedKVPair1, segments[1])
-	require.Equal(t, expectedKVPair2, segments[2])
-	require.Equal(t, expectedKVPair3, segments[3])
-	require.Equal(t, expectedBeginBlockResBytes, segments[4])
-}
-
-func testListenDeliverTx1(t *testing.T) {
-	expectedDeliverTxReq1Bytes, err := testMarshaller.Marshal(&testDeliverTxReq1)
-	require.Nil(t, err)
-	expectedDeliverTxRes1Bytes, err := testMarshaller.Marshal(&testDeliverTxRes1)
-	require.Nil(t, err)
-
 	// write state changes
 	testListener1.OnWrite(mockStoreKey1, mockKey1, mockValue1, false)
 	testListener2.OnWrite(mockStoreKey2, mockKey2, mockValue2, false)
-	testListener1.OnWrite(mockStoreKey2, mockKey3, mockValue3, false)
+	testListener2.OnWrite(mockStoreKey2, mockKey3, mockValue3, false)
 
 	// expected KV pairs
-	expectedKVPair1, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair1, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey1.Name(),
 		Key:      mockKey1,
 		Value:    mockValue1,
 		Delete:   false,
 	})
 	require.Nil(t, err)
-	expectedKVPair2, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair2, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey2.Name(),
 		Key:      mockKey2,
 		Value:    mockValue2,
 		Delete:   false,
 	})
 	require.Nil(t, err)
-	expectedKVPair3, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair3, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey2.Name(),
 		Key:      mockKey3,
 		Value:    mockValue3,
 		Delete:   false,
 	})
 	require.Nil(t, err)
+	expectKVPairsStore1 = append(expectKVPairsStore1, expectedKVPair1)
+	expectKVPairsStore2 = append(expectKVPairsStore2, expectedKVPair2, expectedKVPair3)
 
 	// send the ABCI messages
 	err = testStreamingService.ListenDeliverTx(emptyContext, testDeliverTxReq1, testDeliverTxRes1)
 	require.Nil(t, err)
 
-	// load the file, checking that it was created with the expected name
-	fileName := fmt.Sprintf("%s-block-%d-tx-%d", testPrefix, testBeginBlockReq.GetHeader().Height, 0)
-	fileBytes, err := readInFile(fileName)
-	require.Nil(t, err)
-
-	// segment the file into the separate gRPC messages and check the correctness of each
-	segments, err := segmentBytes(fileBytes)
-	require.Nil(t, err)
-	require.Equal(t, 5, len(segments))
-	require.Equal(t, expectedDeliverTxReq1Bytes, segments[0])
-	require.Equal(t, expectedKVPair1, segments[1])
-	require.Equal(t, expectedKVPair2, segments[2])
-	require.Equal(t, expectedKVPair3, segments[3])
-	require.Equal(t, expectedDeliverTxRes1Bytes, segments[4])
-}
-
-func testListenDeliverTx2(t *testing.T) {
-	expectedDeliverTxReq2Bytes, err := testMarshaller.Marshal(&testDeliverTxReq2)
-	require.Nil(t, err)
-	expectedDeliverTxRes2Bytes, err := testMarshaller.Marshal(&testDeliverTxRes2)
-	require.Nil(t, err)
-
 	// write state changes
-	testListener1.OnWrite(mockStoreKey2, mockKey1, mockValue1, false)
-	testListener2.OnWrite(mockStoreKey1, mockKey2, mockValue2, false)
-	testListener1.OnWrite(mockStoreKey2, mockKey3, mockValue3, false)
+	testListener2.OnWrite(mockStoreKey2, mockKey1, mockValue1, false)
+	testListener1.OnWrite(mockStoreKey1, mockKey2, mockValue2, false)
+	testListener2.OnWrite(mockStoreKey2, mockKey3, mockValue3, false)
 
 	// expected KV pairs
-	expectedKVPair1, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair1, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey2.Name(),
 		Key:      mockKey1,
 		Value:    mockValue1,
 		Delete:   false,
 	})
 	require.Nil(t, err)
-	expectedKVPair2, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair2, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey1.Name(),
 		Key:      mockKey2,
 		Value:    mockValue2,
 		Delete:   false,
 	})
 	require.Nil(t, err)
-	expectedKVPair3, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair3, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey2.Name(),
 		Key:      mockKey3,
 		Value:    mockValue3,
 		Delete:   false,
 	})
 	require.Nil(t, err)
+	expectKVPairsStore1 = append(expectKVPairsStore1, expectedKVPair2)
+	expectKVPairsStore2 = append(expectKVPairsStore2, expectedKVPair1, expectedKVPair3)
 
 	// send the ABCI messages
 	err = testStreamingService.ListenDeliverTx(emptyContext, testDeliverTxReq2, testDeliverTxRes2)
 	require.Nil(t, err)
 
-	// load the file, checking that it was created with the expected name
-	fileName := fmt.Sprintf("%s-block-%d-tx-%d", testPrefix, testBeginBlockReq.GetHeader().Height, 1)
-	fileBytes, err := readInFile(fileName)
-	require.Nil(t, err)
-
-	// segment the file into the separate gRPC messages and check the correctness of each
-	segments, err := segmentBytes(fileBytes)
-	require.Nil(t, err)
-	require.Equal(t, 5, len(segments))
-	require.Equal(t, expectedDeliverTxReq2Bytes, segments[0])
-	require.Equal(t, expectedKVPair1, segments[1])
-	require.Equal(t, expectedKVPair2, segments[2])
-	require.Equal(t, expectedKVPair3, segments[3])
-	require.Equal(t, expectedDeliverTxRes2Bytes, segments[4])
-}
-
-func testListenEndBlock(t *testing.T) {
-	expectedEndBlockReqBytes, err := testMarshaller.Marshal(&testEndBlockReq)
-	require.Nil(t, err)
-	expectedEndBlockResBytes, err := testMarshaller.Marshal(&testEndBlockRes)
-	require.Nil(t, err)
-
 	// write state changes
 	testListener1.OnWrite(mockStoreKey1, mockKey1, mockValue1, false)
-	testListener2.OnWrite(mockStoreKey1, mockKey2, mockValue2, false)
-	testListener1.OnWrite(mockStoreKey2, mockKey3, mockValue3, false)
+	testListener1.OnWrite(mockStoreKey1, mockKey2, mockValue2, false)
+	testListener2.OnWrite(mockStoreKey2, mockKey3, mockValue3, false)
 
 	// expected KV pairs
-	expectedKVPair1, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair1, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey1.Name(),
 		Key:      mockKey1,
 		Value:    mockValue1,
 		Delete:   false,
 	})
 	require.Nil(t, err)
-	expectedKVPair2, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair2, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey1.Name(),
 		Key:      mockKey2,
 		Value:    mockValue2,
 		Delete:   false,
 	})
 	require.Nil(t, err)
-	expectedKVPair3, err := testMarshaller.Marshal(&types.StoreKVPair{
+	expectedKVPair3, err = testMarshaller.Marshal(&types.StoreKVPair{
 		StoreKey: mockStoreKey2.Name(),
 		Key:      mockKey3,
 		Value:    mockValue3,
@@ -346,29 +267,57 @@ func testListenEndBlock(t *testing.T) {
 	})
 	require.Nil(t, err)
 
+	expectKVPairsStore1 = append(expectKVPairsStore1, expectedKVPair1, expectedKVPair2)
+	expectKVPairsStore2 = append(expectKVPairsStore2, expectedKVPair3)
+
 	// send the ABCI messages
 	err = testStreamingService.ListenEndBlock(emptyContext, testEndBlockReq, testEndBlockRes)
 	require.Nil(t, err)
 
-	// load the file, checking that it was created with the expected name
-	fileName := fmt.Sprintf("%s-block-%d-end", testPrefix, testEndBlockReq.Height)
-	fileBytes, err := readInFile(fileName)
+	err = testStreamingService.ListenCommit(emptyContext, testCommitRes)
 	require.Nil(t, err)
 
-	// segment the file into the separate gRPC messages and check the correctness of each
-	segments, err := segmentBytes(fileBytes)
+	// load the file, checking that it was created with the expected name
+	metaFileName := fmt.Sprintf("%s-block-%d-meta", testPrefix, testBeginBlockReq.GetHeader().Height)
+	dataFileName := fmt.Sprintf("%s-block-%d-data", testPrefix, testBeginBlockReq.GetHeader().Height)
+	metaFileBytes, err := readInFile(metaFileName)
+	dataFileBytes, err := readInFile(dataFileName)
 	require.Nil(t, err)
-	require.Equal(t, 5, len(segments))
-	require.Equal(t, expectedEndBlockReqBytes, segments[0])
-	require.Equal(t, expectedKVPair1, segments[1])
-	require.Equal(t, expectedKVPair2, segments[2])
-	require.Equal(t, expectedKVPair3, segments[3])
-	require.Equal(t, expectedEndBlockResBytes, segments[4])
+
+	metadata := types.BlockMetadata{
+		RequestBeginBlock:  &testBeginBlockReq,
+		ResponseBeginBlock: &testBeginBlockRes,
+		RequestEndBlock:    &testEndBlockReq,
+		ResponseEndBlock:   &testEndBlockRes,
+		ResponseCommit:     &testCommitRes,
+		DeliverTxs: []*types.BlockMetadata_DeliverTx{
+			{Request: &testDeliverTxReq1, Response: &testDeliverTxRes1},
+			{Request: &testDeliverTxReq2, Response: &testDeliverTxRes2},
+		},
+	}
+	expectedMetadataBytes, err := testMarshaller.Marshal(&metadata)
+	require.Nil(t, err)
+	require.Equal(t, expectedMetadataBytes, metaFileBytes)
+
+	// segment the file into the separate gRPC messages and check the correctness of each
+	segments, err := segmentBytes(dataFileBytes)
+	require.Nil(t, err)
+	require.Equal(t, len(expectKVPairsStore1)+len(expectKVPairsStore2), len(segments))
+	require.Equal(t, expectKVPairsStore1, segments[:len(expectKVPairsStore1)])
+	require.Equal(t, expectKVPairsStore2, segments[len(expectKVPairsStore1):])
 }
 
 func readInFile(name string) ([]byte, error) {
 	path := filepath.Join(testDir, name)
-	return os.ReadFile(path)
+	bz, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	size := sdk.BigEndianToUint64(bz[:8])
+	if len(bz) != int(size)+8 {
+		return nil, errors.New("incomplete file ")
+	}
+	return bz[8:], nil
 }
 
 // segmentBytes returns all of the protobuf messages contained in the byte array as an array of byte arrays