cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge branch 'master' into raulk/badger-index-cache

Browse Source
This commit is contained in:
Raúl Kripalani
2020-12-02 11:46:57 +00:00
parent 44040f0f3f 3196b2cfd5
commit 655aea1018
213 changed files with 20423 additions and 1577 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+9 -1
View File
@@ -26,6 +26,7 @@ import (
"github.com/libp2p/go-libp2p-peerstore/pstoremem"
pubsub "github.com/libp2p/go-libp2p-pubsub"
record "github.com/libp2p/go-libp2p-record"
"github.com/libp2p/go-libp2p/p2p/net/conngater"
"github.com/multiformats/go-multiaddr"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -98,6 +99,7 @@ var (
ConnectionManagerKey = special{9} // Libp2p option
AutoNATSvcKey = special{10} // Libp2p option
BandwidthReporterKey = special{11} // Libp2p option
ConnGaterKey = special{12} // libp2p option
)
type invoke int
@@ -220,6 +222,9 @@ func libp2p() Option {
Override(PstoreAddSelfKeysKey, lp2p.PstoreAddSelfKeys),
Override(StartListeningKey, lp2p.StartListening(config.DefaultFullNode().Libp2p.ListenAddresses)),
Override(new(*conngater.BasicConnectionGater), lp2p.ConnGater),
Override(ConnGaterKey, lp2p.ConnGaterOption),
)
}
@@ -285,7 +290,7 @@ func Online() Option {
Override(new(exchange.Server), exchange.NewServer),
Override(new(*peermgr.PeerMgr), peermgr.NewPeerMgr),
Override(new(dtypes.Graphsync), modules.Graphsync),
Override(new(dtypes.Graphsync), modules.Graphsync(config.DefaultFullNode().Client.SimultaneousTransfers)),
Override(new(*dtypes.MpoolLocker), new(dtypes.MpoolLocker)),
Override(new(*discoveryimpl.Local), modules.NewLocalDiscovery),
Override(new(discovery.PeerResolver), modules.RetrievalResolver),
@@ -460,12 +465,15 @@ func ConfigFullNode(c interface{}) Option {
ipfsMaddr := cfg.Client.IpfsMAddr
return Options(
ConfigCommon(&cfg.Common),
If(cfg.Client.UseIpfs,
Override(new(dtypes.ClientBlockstore), modules.IpfsClientBlockstore(ipfsMaddr, cfg.Client.IpfsOnlineMode)),
If(cfg.Client.IpfsUseForRetrieval,
Override(new(dtypes.ClientRetrievalStoreManager), modules.ClientBlockstoreRetrievalStoreManager),
),
),
Override(new(dtypes.Graphsync), modules.Graphsync(cfg.Client.SimultaneousTransfers)),
If(cfg.Metrics.HeadNotifs,
Override(HeadMetricsKey, metrics.SendHeadNotifs(cfg.Metrics.Nickname)),
),
node/config/def.go
+9 -4
View File
@@ -105,10 +105,11 @@ type Metrics struct {
}
type Client struct {
UseIpfs bool
IpfsOnlineMode bool
IpfsMAddr string
IpfsUseForRetrieval bool
UseIpfs bool
IpfsOnlineMode bool
IpfsMAddr string
IpfsUseForRetrieval bool
SimultaneousTransfers uint64
}
type Wallet struct {
@@ -149,6 +150,7 @@ func defCommon() Common {
}
var DefaultDefaultMaxFee = types.MustParseFIL("0.007")
var DefaultSimultaneousTransfers = uint64(20)
// DefaultFullNode returns the default config
func DefaultFullNode() *FullNode {
@@ -157,6 +159,9 @@ func DefaultFullNode() *FullNode {
Fees: FeeConfig{
DefaultMaxFee: DefaultDefaultMaxFee,
},
Client: Client{
SimultaneousTransfers: DefaultSimultaneousTransfers,
},
}
}
node/impl/client/client.go
+9 -7
View File
@@ -33,9 +33,10 @@ import (
"go.uber.org/fx"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-commp-utils/ffiwrapper"
"github.com/filecoin-project/go-commp-utils/writer"
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-fil-markets/discovery"
"github.com/filecoin-project/go-fil-markets/pieceio"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
rm "github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/shared"
@@ -49,7 +50,6 @@ import (
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/commp"
"github.com/filecoin-project/lotus/markets/utils"
"github.com/filecoin-project/lotus/node/impl/full"
"github.com/filecoin-project/lotus/node/impl/paych"
@@ -642,7 +642,7 @@ func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Addre
func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet, error) {
// Hard-code the sector type to 32GiBV1_1, because:
// - pieceio.GeneratePieceCommitment requires a RegisteredSealProof
// - ffiwrapper.GeneratePieceCIDFromFile requires a RegisteredSealProof
// - commP itself is sector-size independent, with rather low probability of that changing
// ( note how the final rust call is identical for every RegSP type )
// https://github.com/filecoin-project/rust-filecoin-proofs-api/blob/v5.0.0/src/seal.rs#L1040-L1050
@@ -662,7 +662,8 @@ func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet
return nil, err
}
commP, pieceSize, err := pieceio.GeneratePieceCommitment(arbitraryProofType, rdr, uint64(stat.Size()))
pieceReader, pieceSize := padreader.New(rdr, uint64(stat.Size()))
commP, err := ffiwrapper.GeneratePieceCIDFromFile(arbitraryProofType, pieceReader, pieceSize)
if err != nil {
return nil, xerrors.Errorf("computing commP failed: %w", err)
@@ -702,8 +703,8 @@ func (a *API) ClientDealSize(ctx context.Context, root cid.Cid) (api.DataSize, e
func (a *API) ClientDealPieceCID(ctx context.Context, root cid.Cid) (api.DataCIDSize, error) {
dag := merkledag.NewDAGService(blockservice.New(a.CombinedBstore, offline.Exchange(a.CombinedBstore)))
w := &commp.Writer{}
bw := bufio.NewWriterSize(w, int(commp.CommPBuf))
w := &writer.Writer{}
bw := bufio.NewWriterSize(w, int(writer.CommPBuf))
err := car.WriteCar(ctx, dag, []cid.Cid{root}, w)
if err != nil {
@@ -714,7 +715,8 @@ func (a *API) ClientDealPieceCID(ctx context.Context, root cid.Cid) (api.DataCID
return api.DataCIDSize{}, err
}
return w.Sum()
dataCIDSize, err := w.Sum()
return api.DataCIDSize(dataCIDSize), err
}
func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath string) error {
node/impl/common/common.go
+2
View File
@@ -15,6 +15,7 @@ import (
protocol "github.com/libp2p/go-libp2p-core/protocol"
swarm "github.com/libp2p/go-libp2p-swarm"
basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
"github.com/libp2p/go-libp2p/p2p/net/conngater"
ma "github.com/multiformats/go-multiaddr"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -36,6 +37,7 @@ type CommonAPI struct {
RawHost lp2p.RawHost
Host host.Host
Router lp2p.BaseIpfsRouting
ConnGater *conngater.BasicConnectionGater
Reporter metrics.Reporter
Sk *dtypes.ScoreKeeper
ShutdownChan dtypes.ShutdownChan
node/impl/common/conngater.go
+136
View File
@@ -0,0 +1,136 @@
package common
import (
"context"
"net"
"golang.org/x/xerrors"
logging "github.com/ipfs/go-log/v2"
manet "github.com/multiformats/go-multiaddr/net"
"github.com/filecoin-project/lotus/api"
)
var cLog = logging.Logger("conngater")
func (a *CommonAPI) NetBlockAdd(ctx context.Context, acl api.NetBlockList) error {
for _, p := range acl.Peers {
err := a.ConnGater.BlockPeer(p)
if err != nil {
return xerrors.Errorf("error blocking peer %s: %w", p, err)
}
for _, c := range a.Host.Network().ConnsToPeer(p) {
err = c.Close()
if err != nil {
// just log this, don't fail
cLog.Warnf("error closing connection to %s: %s", p, err)
}
}
}
for _, addr := range acl.IPAddrs {
ip := net.ParseIP(addr)
if ip == nil {
return xerrors.Errorf("error parsing IP address %s", addr)
}
err := a.ConnGater.BlockAddr(ip)
if err != nil {
return xerrors.Errorf("error blocking IP address %s: %w", addr, err)
}
for _, c := range a.Host.Network().Conns() {
remote := c.RemoteMultiaddr()
remoteIP, err := manet.ToIP(remote)
if err != nil {
continue
}
if ip.Equal(remoteIP) {
err = c.Close()
if err != nil {
// just log this, don't fail
cLog.Warnf("error closing connection to %s: %s", remoteIP, err)
}
}
}
}
for _, subnet := range acl.IPSubnets {
_, cidr, err := net.ParseCIDR(subnet)
if err != nil {
return xerrors.Errorf("error parsing subnet %s: %w", subnet, err)
}
err = a.ConnGater.BlockSubnet(cidr)
if err != nil {
return xerrors.Errorf("error blocking subunet %s: %w", subnet, err)
}
for _, c := range a.Host.Network().Conns() {
remote := c.RemoteMultiaddr()
remoteIP, err := manet.ToIP(remote)
if err != nil {
continue
}
if cidr.Contains(remoteIP) {
err = c.Close()
if err != nil {
// just log this, don't fail
cLog.Warnf("error closing connection to %s: %s", remoteIP, err)
}
}
}
}
return nil
}
func (a *CommonAPI) NetBlockRemove(ctx context.Context, acl api.NetBlockList) error {
for _, p := range acl.Peers {
err := a.ConnGater.UnblockPeer(p)
if err != nil {
return xerrors.Errorf("error unblocking peer %s: %w", p, err)
}
}
for _, addr := range acl.IPAddrs {
ip := net.ParseIP(addr)
if ip == nil {
return xerrors.Errorf("error parsing IP address %s", addr)
}
err := a.ConnGater.UnblockAddr(ip)
if err != nil {
return xerrors.Errorf("error unblocking IP address %s: %w", addr, err)
}
}
for _, subnet := range acl.IPSubnets {
_, cidr, err := net.ParseCIDR(subnet)
if err != nil {
return xerrors.Errorf("error parsing subnet %s: %w", subnet, err)
}
err = a.ConnGater.UnblockSubnet(cidr)
if err != nil {
return xerrors.Errorf("error unblocking subunet %s: %w", subnet, err)
}
}
return nil
}
func (a *CommonAPI) NetBlockList(ctx context.Context) (result api.NetBlockList, err error) {
result.Peers = a.ConnGater.ListBlockedPeers()
for _, ip := range a.ConnGater.ListBlockedAddrs() {
result.IPAddrs = append(result.IPAddrs, ip.String())
}
for _, subnet := range a.ConnGater.ListBlockedSubnets() {
result.IPSubnets = append(result.IPSubnets, subnet.String())
}
return
}
node/impl/full/state.go
+6 -3
View File
@@ -54,6 +54,7 @@ type StateModuleAPI interface {
StateMinerProvingDeadline(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*dline.Info, error)
StateMinerPower(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error)
StateNetworkVersion(ctx context.Context, key types.TipSetKey) (network.Version, error)
StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error)
StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error)
}
@@ -792,17 +793,19 @@ func (a *StateAPI) StateSectorPreCommitInfo(ctx context.Context, maddr address.A
pci, err := stmgr.PreCommitInfo(ctx, a.StateManager, maddr, n, ts)
if err != nil {
return miner.SectorPreCommitOnChainInfo{}, err
} else if pci == nil {
return miner.SectorPreCommitOnChainInfo{}, xerrors.Errorf("precommit info is not exists")
}
return *pci, err
}
func (a *StateAPI) StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
func (m *StateModule) StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error) {
ts, err := m.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
return stmgr.MinerSectorInfo(ctx, a.StateManager, maddr, n, ts)
return stmgr.MinerSectorInfo(ctx, m.StateManager, maddr, n, ts)
}
func (a *StateAPI) StateSectorExpiration(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorExpiration, error) {
node/impl/full/wallet.go
+1 -1
View File
@@ -56,7 +56,7 @@ func (a *WalletAPI) WalletSignMessage(ctx context.Context, k address.Address, ms
return nil, xerrors.Errorf("serializing message: %w", err)
}
sig, err := a.WalletAPI.WalletSign(ctx, k, mb.Cid().Bytes(), api.MsgMeta{
sig, err := a.WalletAPI.WalletSign(ctx, keyAddr, mb.Cid().Bytes(), api.MsgMeta{
Type: api.MTChainMsg,
Extra: mb.RawData(),
})
node/impl/storminer.go
+30
View File
@@ -37,6 +37,7 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/storage"
"github.com/filecoin-project/lotus/storage/sectorblocks"
sto "github.com/filecoin-project/specs-storage/storage"
)
type StorageMinerAPI struct {
@@ -543,4 +544,33 @@ func (sm *StorageMinerAPI) CreateBackup(ctx context.Context, fpath string) error
return backup(sm.DS, fpath)
}
func (sm *StorageMinerAPI) CheckProvable(ctx context.Context, pp abi.RegisteredPoStProof, sectors []sto.SectorRef, expensive bool) (map[abi.SectorNumber]string, error) {
var rg storiface.RGetter
if expensive {
rg = func(ctx context.Context, id abi.SectorID) (cid.Cid, error) {
si, err := sm.Miner.GetSectorInfo(id.Number)
if err != nil {
return cid.Undef, err
}
if si.CommR == nil {
return cid.Undef, xerrors.Errorf("commr is nil")
}
return *si.CommR, nil
}
}
bad, err := sm.StorageMgr.CheckProvable(ctx, pp, sectors, rg)
if err != nil {
return nil, err
}
var out = make(map[abi.SectorNumber]string)
for sid, err := range bad {
out[sid.Number] = err
}
return out, nil
}
var _ api.StorageMiner = &StorageMinerAPI{}
node/modules/client.go
+1
View File
@@ -137,6 +137,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
dt.OnReady(marketevents.ReadyLogger("client data transfer"))
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
dt.SubscribeToEvents(marketevents.DataTransferLogger)
return dt.Start(ctx)
},
OnStop: func(ctx context.Context) error {
node/modules/graphsync.go
+30 -26
View File
@@ -3,6 +3,7 @@ package modules
import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
"github.com/ipfs/go-graphsync"
graphsyncimpl "github.com/ipfs/go-graphsync/impl"
gsnet "github.com/ipfs/go-graphsync/network"
@@ -13,31 +14,34 @@ import (
)
// Graphsync creates a graphsync instance from the given loader and storer
func Graphsync(mctx helpers.MetricsCtx, lc fx.Lifecycle, clientBs dtypes.ClientBlockstore, chainBs dtypes.ChainBlockstore, h host.Host) (dtypes.Graphsync, error) {
graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
loader := storeutil.LoaderForBlockstore(clientBs)
storer := storeutil.StorerForBlockstore(clientBs)
gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, loader, storer, graphsyncimpl.RejectAllRequestsByDefault())
chainLoader := storeutil.LoaderForBlockstore(chainBs)
chainStorer := storeutil.StorerForBlockstore(chainBs)
err := gs.RegisterPersistenceOption("chainstore", chainLoader, chainStorer)
if err != nil {
return nil, err
func Graphsync(parallelTransfers uint64) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ChainBlockstore, h host.Host) (dtypes.Graphsync, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, clientBs dtypes.ClientBlockstore, chainBs dtypes.ChainBlockstore, h host.Host) (dtypes.Graphsync, error) {
graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
loader := storeutil.LoaderForBlockstore(clientBs)
storer := storeutil.StorerForBlockstore(clientBs)
gs := graphsyncimpl.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, loader, storer, graphsyncimpl.RejectAllRequestsByDefault(), graphsyncimpl.MaxInProgressRequests(parallelTransfers))
chainLoader := storeutil.LoaderForBlockstore(chainBs)
chainStorer := storeutil.StorerForBlockstore(chainBs)
err := gs.RegisterPersistenceOption("chainstore", chainLoader, chainStorer)
if err != nil {
return nil, err
}
gs.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
_, has := requestData.Extension("chainsync")
if has {
// TODO: we should confirm the selector is a reasonable one before we validate
// TODO: this code will get more complicated and should probably not live here eventually
hookActions.ValidateRequest()
hookActions.UsePersistenceOption("chainstore")
}
})
gs.RegisterOutgoingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.OutgoingRequestHookActions) {
_, has := requestData.Extension("chainsync")
if has {
hookActions.UsePersistenceOption("chainstore")
}
})
return gs, nil
}
gs.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
_, has := requestData.Extension("chainsync")
if has {
// TODO: we should confirm the selector is a reasonable one before we validate
// TODO: this code will get more complicated and should probably not live here eventually
hookActions.ValidateRequest()
hookActions.UsePersistenceOption("chainstore")
}
})
gs.RegisterOutgoingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.OutgoingRequestHookActions) {
_, has := requestData.Extension("chainsync")
if has {
hookActions.UsePersistenceOption("chainstore")
}
})
return gs, nil
}
node/modules/lp2p/conngater.go
+17
View File
@@ -0,0 +1,17 @@
package lp2p
import (
"github.com/libp2p/go-libp2p"
"github.com/libp2p/go-libp2p/p2p/net/conngater"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
func ConnGater(ds dtypes.MetadataDS) (*conngater.BasicConnectionGater, error) {
return conngater.NewBasicConnectionGater(ds)
}
func ConnGaterOption(cg *conngater.BasicConnectionGater) (opts Libp2pOpts, err error) {
opts.Opts = append(opts.Opts, libp2p.ConnectionGater(cg))
return
}
node/modules/storageminer.go
+2 -8
View File
@@ -300,6 +300,7 @@ func NewProviderDAGServiceDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.S
dt.OnReady(marketevents.ReadyLogger("provider data transfer"))
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
dt.SubscribeToEvents(marketevents.DataTransferLogger)
return dt.Start(ctx)
},
OnStop: func(ctx context.Context) error {
@@ -423,15 +424,8 @@ func NewStorageAsk(ctx helpers.MetricsCtx, fapi lapi.FullNode, ds dtypes.Metadat
if err != nil {
return nil, err
}
storedAsk, err := storedask.NewStoredAsk(namespace.Wrap(providerDs, datastore.NewKey("/storage-ask")), datastore.NewKey("latest"), spn, address.Address(minerAddress),
return storedask.NewStoredAsk(namespace.Wrap(providerDs, datastore.NewKey("/storage-ask")), datastore.NewKey("latest"), spn, address.Address(minerAddress),
storagemarket.MaxPieceSize(abi.PaddedPieceSize(mi.SectorSize)))
if err != nil {
return nil, err
}
if err != nil {
return storedAsk, err
}
return storedAsk, nil
}
func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
node/node_test.go
+14 -7
View File
@@ -38,17 +38,24 @@ func TestAPIDealFlow(t *testing.T) {
logging.SetLogLevel("sub", "ERROR")
logging.SetLogLevel("storageminer", "ERROR")
blockTime := 10 * time.Millisecond
// For these tests where the block time is artificially short, just use
// a deal start epoch that is guaranteed to be far enough in the future
// so that the deal starts sealing in time
dealStartEpoch := abi.ChainEpoch(2 << 12)
t.Run("TestDealFlow", func(t *testing.T) {
test.TestDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond, false, false)
test.TestDealFlow(t, builder.MockSbBuilder, blockTime, false, false, dealStartEpoch)
})
t.Run("WithExportedCAR", func(t *testing.T) {
test.TestDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond, true, false)
test.TestDealFlow(t, builder.MockSbBuilder, blockTime, true, false, dealStartEpoch)
})
t.Run("TestDoubleDealFlow", func(t *testing.T) {
test.TestDoubleDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond)
test.TestDoubleDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
})
t.Run("TestFastRetrievalDealFlow", func(t *testing.T) {
test.TestFastRetrievalDealFlow(t, builder.MockSbBuilder, 10*time.Millisecond)
test.TestFastRetrievalDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
})
}
@@ -71,15 +78,15 @@ func TestAPIDealFlowReal(t *testing.T) {
})
t.Run("basic", func(t *testing.T) {
test.TestDealFlow(t, builder.Builder, time.Second, false, false)
test.TestDealFlow(t, builder.Builder, time.Second, false, false, 0)
})
t.Run("fast-retrieval", func(t *testing.T) {
test.TestDealFlow(t, builder.Builder, time.Second, false, true)
test.TestDealFlow(t, builder.Builder, time.Second, false, true, 0)
})
t.Run("retrieval-second", func(t *testing.T) {
test.TestSenondDealRetrieval(t, builder.Builder, time.Second)
test.TestSecondDealRetrieval(t, builder.Builder, time.Second)
})
}