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

backup: append checksum

Browse Source
This commit is contained in:
Łukasz Magiera 2020-10-06 01:50:43 +02:00
parent 70d88f226e
commit 8cdf078999
2 changed files with 93 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
lib/backupds/datastore.go
View File

@ -1,6 +1,7 @@
package backupds
import (
"crypto/sha256"
"io"
"sync"
@ -26,58 +27,81 @@ func Wrap(child datastore.Batching) *Datastore {
}
}
// Writes a datastore dump into the provided writer as indefinite length cbor
// array of [key, value] tuples
// Writes a datastore dump into the provided writer as
// [array(*) of [key, value] tuples, checksum]
func (d *Datastore) Backup(out io.Writer) error {
// write indefinite length array header
if _, err := out.Write([]byte{0x9f}); err != nil {
return xerrors.Errorf("writing header: %w", err)
}
d.backupLk.Lock()
defer d.backupLk.Unlock()
log.Info("Starting datastore backup")
defer log.Info("Datastore backup done")
qr, err := d.child.Query(query.Query{})
if err != nil {
return xerrors.Errorf("query: %w", err)
}
defer func() {
if err := qr.Close(); err != nil {
log.Errorf("query close error: %+v", err)
return
}
}()
scratch := make([]byte, 9)
for result := range qr.Next() {
if err := cbg.WriteMajorTypeHeaderBuf(scratch, out, cbg.MajArray, 2); err != nil {
return xerrors.Errorf("writing tuple header: %w", err)
if err := cbg.WriteMajorTypeHeaderBuf(scratch, out, cbg.MajArray, 2); err != nil {
return xerrors.Errorf("writing tuple header: %w", err)
}
hasher := sha256.New()
hout := io.MultiWriter(hasher, out)
// write KVs
{
// write indefinite length array header
if _, err := hout.Write([]byte{0x9f}); err != nil {
return xerrors.Errorf("writing header: %w", err)
}
if err := cbg.WriteMajorTypeHeaderBuf(scratch, out, cbg.MajByteString, uint64(len([]byte(result.Key)))); err != nil {
return xerrors.Errorf("writing key header: %w", err)
d.backupLk.Lock()
defer d.backupLk.Unlock()
log.Info("Starting datastore backup")
defer log.Info("Datastore backup done")
qr, err := d.child.Query(query.Query{})
if err != nil {
return xerrors.Errorf("query: %w", err)
}
defer func() {
if err := qr.Close(); err != nil {
log.Errorf("query close error: %+v", err)
return
}
}()
for result := range qr.Next() {
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajArray, 2); err != nil {
return xerrors.Errorf("writing tuple header: %w", err)
}
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajByteString, uint64(len([]byte(result.Key)))); err != nil {
return xerrors.Errorf("writing key header: %w", err)
}
if _, err := hout.Write([]byte(result.Key)[:]); err != nil {
return xerrors.Errorf("writing key: %w", err)
}
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajByteString, uint64(len(result.Value))); err != nil {
return xerrors.Errorf("writing value header: %w", err)
}
if _, err := hout.Write(result.Value[:]); err != nil {
return xerrors.Errorf("writing value: %w", err)
}
}
if _, err := out.Write([]byte(result.Key)[:]); err != nil {
return xerrors.Errorf("writing key: %w", err)
}
if err := cbg.WriteMajorTypeHeaderBuf(scratch, out, cbg.MajByteString, uint64(len(result.Value))); err != nil {
return xerrors.Errorf("writing value header: %w", err)
}
if _, err := out.Write(result.Value[:]); err != nil {
return xerrors.Errorf("writing value: %w", err)
// array break
if _, err := hout.Write([]byte{0xff}); err != nil {
return xerrors.Errorf("writing array 'break': %w", err)
}
}
// array break
if _, err := out.Write([]byte{0xff}); err != nil {
return xerrors.Errorf("writing array 'break': %w", err)
// Write the checksum
{
sum := hasher.Sum(nil)
if err := cbg.WriteMajorTypeHeaderBuf(scratch, hout, cbg.MajByteString, uint64(len(sum))); err != nil {
return xerrors.Errorf("writing checksum header: %w", err)
}
if _, err := hout.Write(sum[:]); err != nil {
return xerrors.Errorf("writing checksum: %w", err)
}
}
return nil

30
lib/backupds/read.go
View File

@ -1,6 +1,8 @@
package backupds
import (
"bytes"
"crypto/sha256"
"io"
"github.com/ipfs/go-datastore"
@ -15,12 +17,23 @@ func ReadBackup(r io.Reader, cb func(key datastore.Key, value []byte) error) err
return xerrors.Errorf("reading array header: %w", err)
}
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)
if _, err := hr.Read(scratch[:1]); err != nil {
return xerrors.Errorf("reading array header: %w", err)
}
if scratch[0] != 0x9f {
return xerrors.Errorf("expected indefinite length array header byte 0x9f, got %x", scratch[0])
}
for {
if _, err := r.Read(scratch[:1]); err != nil {
if _, err := hr.Read(scratch[:1]); err != nil {
return xerrors.Errorf("reading tuple header: %w", err)
}
@ -32,13 +45,13 @@ func ReadBackup(r io.Reader, cb func(key datastore.Key, value []byte) error) err
return xerrors.Errorf("expected array(2) header 0x82, got %x", scratch[0])
}
keyb, err := cbg.ReadByteArray(r, 1<<40)
keyb, err := cbg.ReadByteArray(hr, 1<<40)
if err != nil {
return xerrors.Errorf("reading key: %w", err)
}
key := datastore.NewKey(string(keyb))
value, err := cbg.ReadByteArray(r, 1<<40)
value, err := cbg.ReadByteArray(hr, 1<<40)
if err != nil {
return xerrors.Errorf("reading value: %w", err)
}
@ -48,6 +61,17 @@ func ReadBackup(r io.Reader, cb func(key datastore.Key, value []byte) error) err
}
}
sum := hasher.Sum(nil)
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)
}
return nil
}