Large refactor of the datastore layer

* introduce new storage interface that does not perform any
  validation on the data
* add BlobName struct encapsulating blob names, blob names
  encode information about blob type and its hash
* introduce blob type handler interface to deal with different
  blob types, currently only static blobs are supported
* simplify datastore interface - methods do not return readers
  or writers anymore, instead those expect input or output
  stream to be given as an argument

Author: byo

.gitignore

@@ -25,3 +25,5 @@ _testmain.go
 
 coverage.out
 *.coverprofile
+
+*.log

.vscode/settings.json

@@ -1,4 +1,15 @@
 {
     "go.autocompleteUnimportedPackages": true,
-    "go.coverOnSave": true
+    "go.coverOnSave": true,
+    "cSpell.words": [
+        "blobtype",
+        "blobtypes",
+        "bmap",
+        "cinode",
+        "ciphertext",
+        "Hasher",
+        "coverprofile",
+        "goveralls",
+        "shogo"
+    ]
 }

datastore/blob_name.go

@@ -0,0 +1,67 @@
+package datastore
+
+import (
+	"errors"
+
+	base58 "github.com/jbenet/go-base58"
+)
+
+var (
+	ErrInvalidBlobName = errors.New("invalid blob name")
+)
+
+// BlobName is used to identify blobs.
+// Internally it is a single array of bytes that represents
+// both the type of the blob and internal hash used to create that blob.
+// The type of the blob is not stored directly. Instead it is mixed
+// with the hash of the blob to make sure that all bytes in the blob name
+// are randomly distributed.
+type BlobName []byte
+
+// BlobNameFromHashAndType generates the name of a blob from some hash (e.g. sha256 of blob's content)
+// and given blob type
+func BlobNameFromHashAndType(hash []byte, t BlobType) (BlobName, error) {
+	if len(hash) == 0 {
+		return nil, ErrInvalidBlobName
+	}
+
+	ret := make([]byte, len(hash)+1)
+
+	copy(ret[1:], hash)
+
+	scrambledTypeByte := byte(t)
+	for _, b := range hash {
+		scrambledTypeByte ^= b
+	}
+	ret[0] = scrambledTypeByte
+
+	return BlobName(ret), nil
+}
+
+// BlobNameFromString decodes base58-encoded string into blob name
+func BlobNameFromString(s string) (BlobName, error) {
+	decoded := base58.Decode(s)
+	if len(decoded) == 0 {
+		return nil, ErrInvalidBlobName
+	}
+	return BlobName(decoded), nil
+}
+
+// Returns base58-encoded blob name
+func (b BlobName) String() string {
+	return base58.Encode(b)
+}
+
+// Extracts hash from blob name
+func (b BlobName) Hash() []byte {
+	return b[1:]
+}
+
+// Extracts blob type from the name
+func (b BlobName) Type() BlobType {
+	ret := byte(0)
+	for _, by := range b {
+		ret ^= by
+	}
+	return BlobType(ret)
+}

datastore/blob_name_test.go

@@ -0,0 +1,44 @@
+package datastore
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBlobName(t *testing.T) {
+	for _, h := range [][]byte{
+		{0}, {1}, {2}, {0xFE}, {0xFF},
+		{0, 0, 0, 0},
+		{1, 2, 3, 4, 5, 6},
+		sha256.New().Sum(nil),
+	} {
+		for _, bt := range []BlobType{0, 1, 2, 0xFE, 0xFF} {
+			t.Run(fmt.Sprintf("%v:%v", bt, h), func(t *testing.T) {
+				bn, err := BlobNameFromHashAndType(h, bt)
+				assert.NoError(t, err)
+				assert.NotEmpty(t, bn)
+				assert.Greater(t, len(bn), len(h))
+				assert.Equal(t, h, bn.Hash())
+				assert.Equal(t, bt, bn.Type())
+
+				s := bn.String()
+				bn2, err := BlobNameFromString(s)
+				require.NoError(t, err)
+				require.Equal(t, bn, bn2)
+			})
+		}
+	}
+
+	_, err := BlobNameFromString("!@#")
+	require.ErrorIs(t, err, ErrInvalidBlobName)
+
+	_, err = BlobNameFromString("")
+	require.ErrorIs(t, err, ErrInvalidBlobName)
+
+	_, err = BlobNameFromHashAndType(nil, 0x00)
+	require.ErrorIs(t, err, ErrInvalidBlobName)
+}

datastore/blob_type_handler.go

@@ -0,0 +1,14 @@
+package datastore
+
+import "io"
+
+type BlobType byte
+
+type BlobTypeHandler interface {
+	// Type returns the type of blobs supported by this handler
+	Type() BlobType
+
+	Ingest(hash []byte, current, update io.Reader, result io.Writer) error
+
+	Validate(hash []byte, data io.Reader, validated io.Writer) error
+}

datastore/blob_type_handlers_list.go

@@ -0,0 +1,24 @@
+package datastore
+
+import (
+	"errors"
+	"fmt"
+)
+
+var handlers = []BlobTypeHandler{
+	NewStaticBlobHandlerSha256(),
+}
+
+var (
+	ErrUnknownBlobType = errors.New("unknown blob type")
+)
+
+func handlerForType(t BlobType) (BlobTypeHandler, error) {
+	for _, h := range handlers {
+		if h.Type() == t {
+			return h, nil
+		}
+	}
+
+	return nil, fmt.Errorf("%w (%d)", ErrUnknownBlobType, t)
+}

datastore/blob_type_static.go

@@ -0,0 +1,52 @@
+package datastore
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"hash"
+	"io"
+)
+
+var (
+	ErrInvalidStaticBlobHash = fmt.Errorf("%w: invalid static blob hash", ErrValidationFailed)
+)
+
+const (
+	blobTypeStatic BlobType = 0x00
+)
+
+type staticBlobHandler struct {
+	blobType  BlobType
+	newHasher func() hash.Hash
+}
+
+func (h staticBlobHandler) Type() BlobType {
+	return h.blobType
+}
+
+func (h staticBlobHandler) Ingest(hash []byte, current, update io.Reader, w io.Writer) error {
+	// TODO: Optimize - at this point we are not required to ingest the update stream
+	//       so if the current stream is valid (hash correct hash), we should cancel the
+	//		 ingestion (but don't return an error, instead say that update is not needed)
+	return h.Validate(hash, update, w)
+}
+
+func (h staticBlobHandler) Validate(hash []byte, data io.Reader, validated io.Writer) error {
+	_, err := io.Copy(
+		validated,
+		NewChecksumReader(
+			data,
+			h.newHasher(),
+			hash,
+			ErrInvalidStaticBlobHash,
+		),
+	)
+	return err
+}
+
+func NewStaticBlobHandlerSha256() BlobTypeHandler {
+	return staticBlobHandler{
+		blobType:  blobTypeStatic,
+		newHasher: sha256.New,
+	}
+}

datastore/blob_type_static_test.go

@@ -0,0 +1,117 @@
+package datastore
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"errors"
+	"io"
+	"testing"
+	"testing/iotest"
+
+	"github.com/stretchr/testify/require"
+)
+
+func getUpdatedBlob(bt BlobTypeHandler, hash []byte, current, update []byte) ([]byte, error) {
+	var currentReader io.Reader
+	if current != nil {
+		currentReader = bytes.NewReader(current)
+	}
+
+	return getUpdatedBlobWithCurrentReader(bt, hash, currentReader, update)
+}
+
+func getUpdatedBlobWithCurrentReader(bt BlobTypeHandler, hash []byte, currentReader io.Reader, update []byte) ([]byte, error) {
+
+	output := bytes.NewBuffer(nil)
+
+	err := bt.Ingest(hash[:], currentReader, bytes.NewReader(update), output)
+	if err != nil {
+		return nil, err
+	}
+
+	return output.Bytes(), nil
+}
+
+func TestStaticBlobHandler(t *testing.T) {
+	bt := NewStaticBlobHandlerSha256()
+
+	require.EqualValues(t, bt.Type(), 0x00)
+
+	t.Run("ingest a new correct blob", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(data)
+
+		dataBack, err := getUpdatedBlob(bt, hash[:], nil, data)
+		require.NoError(t, err)
+		require.Equal(t, data, dataBack)
+	})
+
+	t.Run("ingest a new incorrect blob - hash of wrong data", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(append(data, 1))
+
+		dataBack, err := getUpdatedBlob(bt, hash[:], nil, data)
+		require.ErrorIs(t, err, ErrInvalidStaticBlobHash)
+		require.ErrorIs(t, err, ErrValidationFailed)
+		require.Nil(t, dataBack)
+	})
+
+	t.Run("ingest a new incorrect blob - hash size mismatch", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(data)
+
+		dataBack, err := getUpdatedBlob(bt, hash[:len(hash)-1], nil, data)
+		require.ErrorIs(t, err, ErrInvalidStaticBlobHash)
+		require.ErrorIs(t, err, ErrValidationFailed)
+		require.Nil(t, dataBack)
+	})
+
+	t.Run("ingest a correct update", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(data)
+
+		dataBack, err := getUpdatedBlob(bt, hash[:], data, data)
+		require.NoError(t, err)
+		require.Equal(t, data, dataBack)
+	})
+
+	t.Run("ingest an incorrect update - hash of wrong data", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(data)
+
+		_, err := getUpdatedBlob(bt, hash[:], data, append(data, 1))
+		require.ErrorIs(t, err, ErrInvalidStaticBlobHash)
+		require.ErrorIs(t, err, ErrValidationFailed)
+	})
+
+	t.Run("ingest a new incorrect blob - hash size mismatch", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(data)
+
+		dataBack, err := getUpdatedBlob(bt, hash[:len(hash)-1], data, data)
+		require.ErrorIs(t, err, ErrInvalidStaticBlobHash)
+		require.ErrorIs(t, err, ErrValidationFailed)
+		require.Nil(t, dataBack)
+	})
+
+	t.Run("recover from a broken local data - hash mismatch", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(data)
+
+		dataBack, err := getUpdatedBlob(bt, hash[:], append(data, 1), data)
+		require.NoError(t, err)
+		require.Equal(t, data, dataBack)
+	})
+
+	t.Run("recover from a broken local data - read error", func(t *testing.T) {
+		data := []byte("hello world!")
+		hash := sha256.Sum256(data)
+
+		dataBack, err := getUpdatedBlobWithCurrentReader(
+			bt, hash[:], iotest.ErrReader(errors.New("err")), data,
+		)
+		require.NoError(t, err)
+		require.Equal(t, data, dataBack)
+	})
+
+}

datastore/checksumreader.go

@@ -0,0 +1,38 @@
+package datastore
+
+import (
+	"crypto/subtle"
+	"errors"
+	"hash"
+	"io"
+)
+
+type checksumReader struct {
+	src               io.Reader
+	hasher            hash.Hash
+	expectedHash      []byte
+	hashMismatchError error
+}
+
+func (cr *checksumReader) Read(b []byte) (int, error) {
+	n, err := cr.src.Read(b)
+	cr.hasher.Write(b[:n])
+
+	if errors.Is(err, io.EOF) {
+		calculatedHash := cr.hasher.Sum(nil)
+		if subtle.ConstantTimeCompare(cr.expectedHash, calculatedHash) != 1 {
+			return n, cr.hashMismatchError
+		}
+	}
+
+	return n, err
+}
+
+func NewChecksumReader(src io.Reader, hasher hash.Hash, expectedHash []byte, hashMismatchError error) io.Reader {
+	return &checksumReader{
+		src:               src,
+		hasher:            hasher,
+		expectedHash:      expectedHash,
+		hashMismatchError: hashMismatchError,
+	}
+}