write messages to a temp blockstore when validating

chain/sync.go

@@ -323,25 +323,35 @@ func (syncer *Syncer) ValidateMsgMeta(fblk *types.FullBlock) error {
 		return xerrors.Errorf("block %s has too many messages (%d)", fblk.Header.Cid(), msgc)
 	}
 
-	// Collect the CIDs of both types of messages separately: BLS and Secpk.
-	var bcids, scids []cid.Cid
-	for _, m := range fblk.BlsMessages {
-		bcids = append(bcids, m.Cid())
-	}
-
-	for _, m := range fblk.SecpkMessages {
-		scids = append(scids, m.Cid())
-	}
-
 	// TODO: IMPORTANT(GARBAGE). These message puts and the msgmeta
 	// computation need to go into the 'temporary' side of the blockstore when
 	// we implement that
-	blockstore := syncer.store.Blockstore()
 
-	bs := cbor.NewCborStore(blockstore)
+	// We use a temporary bstore here to avoid writing intermediate pieces
+	// into the blockstore.
+	blockstore := bstore.NewTemporary()
+	cst := cbor.NewCborStore(blockstore)
+
+	var bcids, scids []cid.Cid
+
+	for _, m := range fblk.BlsMessages {
+		c, err := store.PutMessage(blockstore, m)
+		if err != nil {
+			return xerrors.Errorf("putting bls message to blockstore after msgmeta computation: %w", err)
+		}
+		bcids = append(bcids, c)
+	}
+
+	for _, m := range fblk.SecpkMessages {
+		c, err := store.PutMessage(blockstore, m)
+		if err != nil {
+			return xerrors.Errorf("putting bls message to blockstore after msgmeta computation: %w", err)
+		}
+		scids = append(scids, c)
+	}
 
 	// Compute the root CID of the combined message trie.
-	smroot, err := computeMsgMeta(bs, bcids, scids)
+	smroot, err := computeMsgMeta(cst, bcids, scids)
 	if err != nil {
 		return xerrors.Errorf("validating msgmeta, compute failed: %w", err)
 	}
@@ -351,21 +361,8 @@ func (syncer *Syncer) ValidateMsgMeta(fblk *types.FullBlock) error {
 		return xerrors.Errorf("messages in full block did not match msgmeta root in header (%s != %s)", fblk.Header.Messages, smroot)
 	}
 
-	for _, m := range fblk.BlsMessages {
+	// Finally, flush.
-		_, err := store.PutMessage(blockstore, m)
+	return vm.Copy(context.TODO(), blockstore, syncer.store.Blockstore(), smroot)
-		if err != nil {
-			return xerrors.Errorf("putting bls message to blockstore after msgmeta computation: %w", err)
-		}
-	}
-
-	for _, m := range fblk.SecpkMessages {
-		_, err := store.PutMessage(blockstore, m)
-		if err != nil {
-			return xerrors.Errorf("putting bls message to blockstore after msgmeta computation: %w", err)
-		}
-	}
-
-	return nil
 }
 
 func (syncer *Syncer) LocalPeer() peer.ID {
@@ -1064,8 +1061,7 @@ func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock
 		return err
 	}
 
-	cst := cbor.NewCborStore(syncer.store.Blockstore())
+	st, err := state.LoadStateTree(syncer.store.Store(ctx), stateroot)
-	st, err := state.LoadStateTree(cst, stateroot)
 	if err != nil {
 		return xerrors.Errorf("failed to load base state tree: %w", err)
 	}
@@ -1111,21 +1107,28 @@ func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock
 		return nil
 	}
 
-	store := adt0.WrapStore(ctx, cst)
+	// Validate message arrays in a temporary blockstore.
+	tmpbs := bstore.NewTemporary()
+	tmpstore := adt0.WrapStore(ctx, cbor.NewCborStore(tmpbs))
 
-	bmArr := adt0.MakeEmptyArray(store)
+	bmArr := adt0.MakeEmptyArray(tmpstore)
 	for i, m := range b.BlsMessages {
 		if err := checkMsg(m); err != nil {
 			return xerrors.Errorf("block had invalid bls message at index %d: %w", i, err)
 		}
 
-		c := cbg.CborCid(m.Cid())
+		c, err := store.PutMessage(tmpbs, m)
-		if err := bmArr.Set(uint64(i), &c); err != nil {
+		if err != nil {
+			return xerrors.Errorf("failed to store message %s: %w", m.Cid(), err)
+		}
+
+		k := cbg.CborCid(c)
+		if err := bmArr.Set(uint64(i), &k); err != nil {
 			return xerrors.Errorf("failed to put bls message at index %d: %w", i, err)
 		}
 	}
 
-	smArr := adt0.MakeEmptyArray(store)
+	smArr := adt0.MakeEmptyArray(tmpstore)
 	for i, m := range b.SecpkMessages {
 		if err := checkMsg(m); err != nil {
 			return xerrors.Errorf("block had invalid secpk message at index %d: %w", i, err)
@@ -1142,8 +1145,12 @@ func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock
 			return xerrors.Errorf("secpk message %s has invalid signature: %w", m.Cid(), err)
 		}
 
-		c := cbg.CborCid(m.Cid())
+		c, err := store.PutMessage(tmpbs, m)
-		if err := smArr.Set(uint64(i), &c); err != nil {
+		if err != nil {
+			return xerrors.Errorf("failed to store message %s: %w", m.Cid(), err)
+		}
+		k := cbg.CborCid(c)
+		if err := smArr.Set(uint64(i), &k); err != nil {
 			return xerrors.Errorf("failed to put secpk message at index %d: %w", i, err)
 		}
 	}
@@ -1158,7 +1165,7 @@ func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock
 		return err
 	}
 
-	mrcid, err := cst.Put(ctx, &types.MsgMeta{
+	mrcid, err := tmpstore.Put(ctx, &types.MsgMeta{
 		BlsMessages:   bmroot,
 		SecpkMessages: smroot,
 	})
@@ -1170,7 +1177,8 @@ func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock
 		return fmt.Errorf("messages didnt match message root in header")
 	}
 
-	return nil
+	// Finally, flush.
+	return vm.Copy(ctx, tmpbs, syncer.store.Blockstore(), mrcid)
 }
 
 func (syncer *Syncer) verifyBlsAggregate(ctx context.Context, sig *crypto.Signature, msgs []cid.Cid, pubks [][]byte) error {