feat: state: Fast migration for v15

blockstore/autobatch.go

@@ -0,0 +1,208 @@
+package blockstore
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"golang.org/x/xerrors"
+
+	block "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+)
+
+// autolog is a logger for the autobatching blockstore. It is subscoped from the
+// blockstore logger.
+var autolog = log.Named("auto")
+
+type AutobatchBlockstore struct {
+	// TODO: drop if memory consumption is too high
+	addedCids map[cid.Cid]struct{}
+
+	bufferedLk          sync.Mutex
+	bufferedBlksOrdered []block.Block
+	bufferedBlksMap     map[cid.Cid]block.Block
+
+	flushingLk      sync.Mutex
+	flushingBlksMap map[cid.Cid]block.Block
+
+	flushCh         chan struct{}
+	flushErr        error
+	flushRetryDelay time.Duration
+	flushCtx        context.Context
+	shutdownCh      chan struct{}
+
+	backingBs Blockstore
+
+	bufferCapacity int
+	bufferSize     int
+}
+
+func NewAutobatch(ctx context.Context, backingBs Blockstore, bufferCapacity int) *AutobatchBlockstore {
+	bs := &AutobatchBlockstore{
+		addedCids:       make(map[cid.Cid]struct{}),
+		backingBs:       backingBs,
+		bufferCapacity:  bufferCapacity,
+		bufferedBlksMap: make(map[cid.Cid]block.Block),
+		flushingBlksMap: make(map[cid.Cid]block.Block),
+		flushCtx:        ctx,
+		flushCh:         make(chan struct{}, 1),
+		// could be made configable
+		flushRetryDelay: time.Second * 5,
+	}
+
+	go bs.flushWorker()
+
+	return bs
+}
+
+func (bs *AutobatchBlockstore) Put(ctx context.Context, blk block.Block) error {
+	bs.bufferedLk.Lock()
+	_, ok := bs.addedCids[blk.Cid()]
+	if !ok {
+		bs.addedCids[blk.Cid()] = struct{}{}
+		bs.bufferedBlksOrdered = append(bs.bufferedBlksOrdered, blk)
+		bs.bufferedBlksMap[blk.Cid()] = blk
+		bs.bufferSize += len(blk.RawData())
+		if bs.bufferSize >= bs.bufferCapacity {
+			// signal that a flush is appropriate, may be ignored
+			select {
+			case bs.flushCh <- struct{}{}:
+			default:
+				// do nothing
+			}
+		}
+	}
+	bs.bufferedLk.Unlock()
+	return nil
+}
+
+func (bs *AutobatchBlockstore) flushWorker() {
+	for {
+		select {
+		case <-bs.flushCh:
+			putErr := bs.doFlush(bs.flushCtx)
+			for putErr != nil {
+				select {
+				case <-bs.shutdownCh:
+					bs.flushErr = putErr
+					return
+				default:
+					autolog.Errorf("FLUSH ERRORED: %w, retrying in %v", putErr, bs.flushRetryDelay)
+					time.Sleep(bs.flushRetryDelay)
+					putErr = bs.doFlush(bs.flushCtx)
+				}
+			}
+		case <-bs.shutdownCh:
+			return
+		}
+	}
+}
+
+func (bs *AutobatchBlockstore) doFlush(ctx context.Context) error {
+	bs.bufferedLk.Lock()
+	bs.flushingLk.Lock()
+	// We do NOT clear addedCids here, because its purpose is to expedite Puts
+	flushingBlksOrdered := bs.bufferedBlksOrdered
+	bs.flushingBlksMap = bs.bufferedBlksMap
+	bs.bufferedBlksOrdered = []block.Block{}
+	bs.bufferedBlksMap = make(map[cid.Cid]block.Block)
+	bs.bufferedLk.Unlock()
+	bs.flushingLk.Unlock()
+	return bs.backingBs.PutMany(ctx, flushingBlksOrdered)
+}
+
+func (bs *AutobatchBlockstore) Shutdown(ctx context.Context) error {
+	// request one last flush of the worker
+	bs.flushCh <- struct{}{}
+	// shutdown the flush worker
+	bs.shutdownCh <- struct{}{}
+	// if it ever errored, this method fails
+	if bs.flushErr != nil {
+		return xerrors.Errorf("flushWorker errored: %w", bs.flushErr)
+	}
+
+	// one last flush in case it's needed
+	return bs.doFlush(ctx)
+}
+
+func (bs *AutobatchBlockstore) Get(ctx context.Context, c cid.Cid) (block.Block, error) {
+	// may seem backward to check the backingBs first, but that is the likeliest case
+	blk, err := bs.backingBs.Get(ctx, c)
+	if err == nil {
+		return blk, nil
+	}
+
+	if err != ErrNotFound {
+		return blk, err
+	}
+
+	bs.flushingLk.Lock()
+	v, ok := bs.flushingBlksMap[c]
+	bs.flushingLk.Unlock()
+	if ok {
+		return v, nil
+	}
+
+	bs.bufferedLk.Lock()
+	v, ok = bs.bufferedBlksMap[c]
+	bs.bufferedLk.Unlock()
+	if ok {
+		return v, nil
+	}
+
+	return nil, ErrNotFound
+}
+
+func (bs *AutobatchBlockstore) DeleteBlock(context.Context, cid.Cid) error {
+	return xerrors.New("deletion is unsupported")
+}
+
+func (bs *AutobatchBlockstore) DeleteMany(ctx context.Context, cids []cid.Cid) error {
+	// see note in DeleteBlock()
+	return xerrors.New("deletion is unsupported")
+}
+
+func (bs *AutobatchBlockstore) Has(ctx context.Context, c cid.Cid) (bool, error) {
+	_, err := bs.Get(ctx, c)
+	if err == nil {
+		return true, nil
+	}
+	if err == ErrNotFound {
+		return false, nil
+	}
+
+	return false, err
+}
+
+func (bs *AutobatchBlockstore) GetSize(ctx context.Context, c cid.Cid) (int, error) {
+	blk, err := bs.Get(ctx, c)
+	if err != nil {
+		return 0, err
+	}
+
+	return len(blk.RawData()), nil
+}
+
+func (bs *AutobatchBlockstore) PutMany(ctx context.Context, blks []block.Block) error {
+	for _, blk := range blks {
+		if err := bs.Put(ctx, blk); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (bs *AutobatchBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
+	return nil, xerrors.New("unsupported")
+
+}
+
+func (bs *AutobatchBlockstore) HashOnRead(enabled bool) {
+	bs.backingBs.HashOnRead(enabled)
+}
+
+func (bs *AutobatchBlockstore) View(ctx context.Context, cid cid.Cid, callback func([]byte) error) error {
+	return xerrors.New("unsupported")
+}

blockstore/autobatch_test.go

@@ -0,0 +1,33 @@
+package blockstore
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAutobatchBlockstore(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	ab := NewAutobatch(ctx, NewMemory(), len(b0.RawData())+len(b1.RawData())-1)
+
+	require.NoError(t, ab.Put(ctx, b0))
+	require.NoError(t, ab.Put(ctx, b1))
+	require.NoError(t, ab.Put(ctx, b2))
+
+	ab.Flush(ctx)
+
+	v0, err := ab.Get(ctx, b0.Cid())
+	require.NoError(t, err)
+	require.Equal(t, b0.RawData(), v0.RawData())
+
+	v1, err := ab.Get(ctx, b1.Cid())
+	require.NoError(t, err)
+	require.Equal(t, b1.RawData(), v1.RawData())
+
+	v2, err := ab.Get(ctx, b2.Cid())
+	require.NoError(t, err)
+	require.Equal(t, b2.RawData(), v2.RawData())
+}

chain/actors/builtin/paych/message.go.template

@@ -39,7 +39,11 @@ func (m message{{.v}}) Create(to address.Address, initialAmount abi.TokenAmount)
 
 func (m message{{.v}}) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych{{.v}}.UpdateChannelStateParams{
+	{{if (ge .v 7)}}
+	    Sv:     toV{{.v}}SignedVoucher(*sv),
+	{{else}}
 		Sv:     *sv,
+	{{end}}
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/message0.go

@@ -39,7 +39,9 @@ func (m message0) Create(to address.Address, initialAmount abi.TokenAmount) (*ty
 
 func (m message0) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych0.UpdateChannelStateParams{
-		Sv:     *sv,
+
+		Sv: *sv,
+
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/message2.go

@@ -39,7 +39,9 @@ func (m message2) Create(to address.Address, initialAmount abi.TokenAmount) (*ty
 
 func (m message2) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych2.UpdateChannelStateParams{
-		Sv:     *sv,
+
+		Sv: *sv,
+
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/message3.go

@@ -39,7 +39,9 @@ func (m message3) Create(to address.Address, initialAmount abi.TokenAmount) (*ty
 
 func (m message3) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych3.UpdateChannelStateParams{
-		Sv:     *sv,
+
+		Sv: *sv,
+
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/message4.go

@@ -39,7 +39,9 @@ func (m message4) Create(to address.Address, initialAmount abi.TokenAmount) (*ty
 
 func (m message4) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych4.UpdateChannelStateParams{
-		Sv:     *sv,
+
+		Sv: *sv,
+
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/message5.go

@@ -39,7 +39,9 @@ func (m message5) Create(to address.Address, initialAmount abi.TokenAmount) (*ty
 
 func (m message5) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych5.UpdateChannelStateParams{
-		Sv:     *sv,
+
+		Sv: *sv,
+
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/message6.go

@@ -39,7 +39,9 @@ func (m message6) Create(to address.Address, initialAmount abi.TokenAmount) (*ty
 
 func (m message6) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych6.UpdateChannelStateParams{
-		Sv:     *sv,
+
+		Sv: *sv,
+
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/message7.go

@@ -39,7 +39,9 @@ func (m message7) Create(to address.Address, initialAmount abi.TokenAmount) (*ty
 
 func (m message7) Update(paych address.Address, sv *SignedVoucher, secret []byte) (*types.Message, error) {
 	params, aerr := actors.SerializeParams(&paych7.UpdateChannelStateParams{
-		Sv:     *sv,
+
+		Sv: toV7SignedVoucher(*sv),
+
 		Secret: secret,
 	})
 	if aerr != nil {

chain/actors/builtin/paych/state.go.template

@@ -112,3 +112,21 @@ func (ls *laneState{{.v}}) Redeemed() (big.Int, error) {
 func (ls *laneState{{.v}}) Nonce() (uint64, error) {
 	return ls.LaneState.Nonce, nil
 }
+
+{{if (ge .v 7)}}
+func toV{{.v}}SignedVoucher(sv SignedVoucher) paych{{.v}}.SignedVoucher {
+	return paych{{.v}}.SignedVoucher{
+		ChannelAddr:     sv.ChannelAddr,
+		TimeLockMin:     sv.TimeLockMin,
+		TimeLockMax:     sv.TimeLockMax,
+		SecretHash:      sv.SecretPreimage,
+		Extra:           sv.Extra,
+		Lane:            sv.Lane,
+		Nonce:           sv.Nonce,
+		Amount:          sv.Amount,
+		MinSettleHeight: sv.MinSettleHeight,
+		Merges:          sv.Merges,
+		Signature:       sv.Signature,
+	}
+}
+{{end}}

chain/actors/builtin/paych/v7.go

@@ -112,3 +112,19 @@ func (ls *laneState7) Redeemed() (big.Int, error) {
 func (ls *laneState7) Nonce() (uint64, error) {
 	return ls.LaneState.Nonce, nil
 }
+
+func toV7SignedVoucher(sv SignedVoucher) paych7.SignedVoucher {
+	return paych7.SignedVoucher{
+		ChannelAddr:     sv.ChannelAddr,
+		TimeLockMin:     sv.TimeLockMin,
+		TimeLockMax:     sv.TimeLockMax,
+		SecretHash:      sv.SecretPreimage,
+		Extra:           sv.Extra,
+		Lane:            sv.Lane,
+		Nonce:           sv.Nonce,
+		Amount:          sv.Amount,
+		MinSettleHeight: sv.MinSettleHeight,
+		Merges:          sv.Merges,
+		Signature:       sv.Signature,
+	}
+}