lotus/lib/backupds/datastore.go

package backupds

import (
	"io"
	"sync"

	logging "github.com/ipfs/go-log/v2"
	cbg "github.com/whyrusleeping/cbor-gen"
	"golang.org/x/xerrors"

	"github.com/ipfs/go-datastore"
	"github.com/ipfs/go-datastore/query"
)

var log = logging.Logger("backupds")

type Datastore struct {
	child datastore.Batching

	backupLk sync.RWMutex
}

func Wrap(child datastore.Batching) *Datastore {
	return &Datastore{
		child: child,
	}
}

// Writes a datastore dump into the provided writer as indefinite length cbor
// array of [key, value] tuples
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.MajByteString, uint64(len([]byte(result.Key)))); err != nil {
			return xerrors.Errorf("writing key header: %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 := out.Write([]byte{0xff}); err != nil {
		return xerrors.Errorf("writing array 'break': %w", err)
	}

	return nil
}

// proxy

func (d *Datastore) Get(key datastore.Key) (value []byte, err error) {
	return d.child.Get(key)
}

func (d *Datastore) Has(key datastore.Key) (exists bool, err error) {
	return d.child.Has(key)
}

func (d *Datastore) GetSize(key datastore.Key) (size int, err error) {
	return d.child.GetSize(key)
}

func (d *Datastore) Query(q query.Query) (query.Results, error) {
	return d.child.Query(q)
}

func (d *Datastore) Put(key datastore.Key, value []byte) error {
	d.backupLk.RLock()
	defer d.backupLk.RUnlock()

	return d.child.Put(key, value)
}

func (d *Datastore) Delete(key datastore.Key) error {
	d.backupLk.RLock()
	defer d.backupLk.RUnlock()

	return d.child.Delete(key)
}

func (d *Datastore) Sync(prefix datastore.Key) error {
	d.backupLk.RLock()
	defer d.backupLk.RUnlock()

	return d.child.Sync(prefix)
}

func (d *Datastore) Close() error {
	d.backupLk.RLock()
	defer d.backupLk.RUnlock()

	return d.child.Close()
}

func (d *Datastore) Batch() (datastore.Batch, error) {
	b, err := d.child.Batch()
	if err != nil {
		return nil, err
	}

	return &bbatch{
		b:   b,
		rlk: d.backupLk.RLocker(),
	}, nil
}

type bbatch struct {
	b   datastore.Batch
	rlk sync.Locker
}

func (b *bbatch) Put(key datastore.Key, value []byte) error {
	return b.b.Put(key, value)
}

func (b *bbatch) Delete(key datastore.Key) error {
	return b.b.Delete(key)
}

func (b *bbatch) Commit() error {
	b.rlk.Lock()
	defer b.rlk.Unlock()

	return b.b.Commit()
}

var _ datastore.Batch = &bbatch{}
var _ datastore.Batching = &Datastore{}