src/server/pkg/hashtree/db.go
func (h *dbHashTree) Deserialize(_r io.Reader) error {
	r := pbutil.NewReader(_r)
	hdr := &BucketHeader{}
	batchSize := 10000

	kvs := make(chan *keyValue, batchSize/10)
	// create cancellable ctx in case bolt writer encounters error
	eg, copyCtx := errgroup.WithContext(context.Background())
	eg.Go(func() error {
		var bucket []byte
		for {
			count := 0
			if err := h.Update(func(tx *bolt.Tx) error {
				if bucket != nil {
					tx.Bucket(bucket).FillPercent = 1
				}
				for kv := range kvs {
					if kv.k == nil {
						bucket = kv.v
						continue
					}
					if err := tx.Bucket(bucket).Put(kv.k, kv.v); err != nil {
						return err
					}
					count++
					if count >= batchSize {
						return nil
					}
				}
				return nil
			}); err != nil || copyCtx.Err() != nil {
				return err // may return nil if copyCtx was closed
			}
			if count <= 0 {
				return nil
			}
		}
	})
	eg.Go(func() error {
		defer close(kvs)
		for {
			hdr.Reset()
			// TODO(msteffen): don't block on Read if copyCtx() is cancelled?
			if err := r.Read(hdr); err != nil {
				if err == io.EOF {
					break
				}
				return err
			}
			bucket := b(hdr.Bucket)
			select {
			case kvs <- &keyValue{nil, bucket}:
			case <-copyCtx.Done():
				return nil
			}
			for {
				_k, err := r.ReadBytes()
				if err != nil {
					return err
				}
				if bytes.Equal(_k, SentinelByte) {
					break
				}
				// we need to make copies of k and v because the memory will be reused
				k := make([]byte, len(_k))
				copy(k, _k)
				_v, err := r.ReadBytes()
				if err != nil {
					return err
				}
				v := make([]byte, len(_v))
				copy(v, _v)
				select {
				case kvs <- &keyValue{k, v}:
				case <-copyCtx.Done():
					return nil
				}
			}
		}
		return nil
	})
	return eg.Wait()
}
func (h *dbHashTree) Deserialize(_r io.Reader) error {
	r := pbutil.NewReader(_r)
	hdr := &BucketHeader{}
	batchSize := 10000

	kvs := make(chan *keyValue, batchSize/10)
	// create cancellable ctx in case bolt writer encounters error
	eg, copyCtx := errgroup.WithContext(context.Background())
	eg.Go(func() error {
		var bucket []byte
		for {
			count := 0
			if err := h.Update(func(tx *bolt.Tx) error {
				if bucket != nil {
					tx.Bucket(bucket).FillPercent = 1
				}
				for kv := range kvs {
					if kv.k == nil {
						bucket = kv.v
						continue
					}
					if err := tx.Bucket(bucket).Put(kv.k, kv.v); err != nil {
						return err
					}
					count++
					if count >= batchSize {
						return nil
					}
				}
				return nil
			}); err != nil || copyCtx.Err() != nil {
				return err // may return nil if copyCtx was closed
			}
			if count <= 0 {
				return nil
			}
		}
	})
	eg.Go(func() error {
		defer close(kvs)
		for {
			hdr.Reset()
			// TODO(msteffen): don't block on Read if copyCtx() is cancelled?
			if err := r.Read(hdr); err != nil {
				if err == io.EOF {
					break
				}
				return err
			}
			bucket := b(hdr.Bucket)
			select {
			case kvs <- &keyValue{nil, bucket}:
			case <-copyCtx.Done():
				return nil
			}
			for {
				_k, err := r.ReadBytes()
				if err != nil {
					return err
				}
				if bytes.Equal(_k, SentinelByte) {
					break
				}
				// we need to make copies of k and v because the memory will be reused
				k := make([]byte, len(_k))
				copy(k, _k)
				_v, err := r.ReadBytes()
				if err != nil {
					return err
				}
				v := make([]byte, len(_v))
				copy(v, _v)
				select {
				case kvs <- &keyValue{k, v}:
				case <-copyCtx.Done():
					return nil
				}
			}
		}
		return nil
	})
	return eg.Wait()
}
// Deserialize deserializes a hashtree.