lotus/lib/backupds/read.go

142 lines
3.2 KiB
Go
Raw Normal View History

2020-10-01 12:36:19 +00:00
package backupds
import (
2020-10-05 23:50:43 +00:00
"bytes"
"crypto/sha256"
2020-10-01 12:36:19 +00:00
"io"
2021-03-09 21:33:01 +00:00
"os"
2020-10-01 12:36:19 +00:00
"github.com/ipfs/go-datastore"
cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors"
)
func ReadBackup(r io.Reader, cb func(key datastore.Key, value []byte) error) error {
scratch := make([]byte, 9)
2021-03-09 21:33:01 +00:00
// read array[2](
2020-10-01 12:36:19 +00:00
if _, err := r.Read(scratch[:1]); err != nil {
return xerrors.Errorf("reading array header: %w", err)
}
2020-10-05 23:50:43 +00:00
if scratch[0] != 0x82 {
return xerrors.Errorf("expected array(2) header byte 0x82, got %x", scratch[0])
}
hasher := sha256.New()
hr := io.TeeReader(r, hasher)
2021-03-09 21:33:01 +00:00
// read array[*](
2020-10-05 23:50:43 +00:00
if _, err := hr.Read(scratch[:1]); err != nil {
return xerrors.Errorf("reading array header: %w", err)
}
2020-10-01 12:36:19 +00:00
if scratch[0] != 0x9f {
return xerrors.Errorf("expected indefinite length array header byte 0x9f, got %x", scratch[0])
}
for {
2020-10-05 23:50:43 +00:00
if _, err := hr.Read(scratch[:1]); err != nil {
2020-10-01 12:36:19 +00:00
return xerrors.Errorf("reading tuple header: %w", err)
}
2021-03-09 21:33:01 +00:00
// close array[*]
2020-10-01 12:36:19 +00:00
if scratch[0] == 0xff {
break
}
2021-03-09 21:33:01 +00:00
// read array[2](key:[]byte, value:[]byte)
2020-10-01 12:36:19 +00:00
if scratch[0] != 0x82 {
return xerrors.Errorf("expected array(2) header 0x82, got %x", scratch[0])
}
2020-10-05 23:50:43 +00:00
keyb, err := cbg.ReadByteArray(hr, 1<<40)
2020-10-01 12:36:19 +00:00
if err != nil {
return xerrors.Errorf("reading key: %w", err)
}
key := datastore.NewKey(string(keyb))
2020-10-05 23:50:43 +00:00
value, err := cbg.ReadByteArray(hr, 1<<40)
2020-10-01 12:36:19 +00:00
if err != nil {
return xerrors.Errorf("reading value: %w", err)
}
if err := cb(key, value); err != nil {
return err
}
}
2020-10-05 23:50:43 +00:00
sum := hasher.Sum(nil)
2021-03-09 21:33:01 +00:00
// read the [32]byte checksum
2020-10-05 23:50:43 +00:00
expSum, err := cbg.ReadByteArray(r, 32)
if err != nil {
return xerrors.Errorf("reading expected checksum: %w", err)
}
if !bytes.Equal(sum, expSum) {
return xerrors.Errorf("checksum didn't match; expected %x, got %x", expSum, sum)
}
2021-03-09 21:33:01 +00:00
// read the log, set of Entry-ies
var ent Entry
bp := cbg.GetPeeker(r)
for {
_, err := bp.ReadByte()
switch err {
case io.EOF, io.ErrUnexpectedEOF:
return nil
case nil:
default:
return xerrors.Errorf("peek log: %w", err)
}
if err := bp.UnreadByte(); err != nil {
return xerrors.Errorf("unread log byte: %w", err)
}
if err := ent.UnmarshalCBOR(bp); err != nil {
switch err {
case io.EOF, io.ErrUnexpectedEOF:
if os.Getenv("LOTUS_ALLOW_TRUNCATED_LOG") == "1" {
panic("handleme; just ignore and tell the caller about the corrupted file") // todo
} else {
return xerrors.Errorf("log entry potentially truncated, set LOTUS_ALLOW_TRUNCATED_LOG=1 to proceed: %w", err)
}
default:
return xerrors.Errorf("unmarshaling log entry: %w", err)
}
}
key := datastore.NewKey(string(ent.Key))
if err := cb(key, ent.Value); err != nil {
return err
}
}
2020-10-01 12:36:19 +00:00
}
2020-10-01 14:22:54 +00:00
func RestoreInto(r io.Reader, dest datastore.Batching) error {
batch, err := dest.Batch()
if err != nil {
return xerrors.Errorf("creating batch: %w", err)
}
err = ReadBackup(r, func(key datastore.Key, value []byte) error {
if err := batch.Put(key, value); err != nil {
return xerrors.Errorf("put key: %w", err)
}
return nil
})
if err != nil {
return xerrors.Errorf("reading backup: %w", err)
}
if err := batch.Commit(); err != nil {
return xerrors.Errorf("committing batch: %w", err)
}
return nil
}