Cac refactor (#1281)

pkg/api/chunk.go

@@ -12,7 +12,7 @@ import (
 	"io/ioutil"
 	"net/http"
 
-	"github.com/ethersphere/bee/pkg/bmtpool"
+	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/netstore"
 
 	"github.com/ethersphere/bee/pkg/jsonhttp"
@@ -75,28 +75,17 @@ func (s *server) chunkUploadHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	hasher := bmtpool.Get()
-	defer bmtpool.Put(hasher)
-
-	err = hasher.SetSpanBytes(data[:swarm.SpanSize])
-	if err != nil {
-		s.logger.Debugf("chunk upload: set span: %v", err)
-		s.logger.Error("chunk upload: span error")
-		jsonhttp.InternalServerError(w, "span error")
-		return
-	}
-
-	_, err = hasher.Write(data[swarm.SpanSize:])
+	chunk, err := cac.NewWithDataSpan(data)
 	if err != nil {
+		s.logger.Debugf("chunk upload: create chunk error: %v", err)
+		s.logger.Error("chunk upload: create chunk error")
+		jsonhttp.InternalServerError(w, "create chunk error")
 		return
 	}
 
-	address := swarm.NewAddress(hasher.Sum(nil))
-	chunk := swarm.NewChunk(address, data)
-
 	seen, err := s.storer.Put(ctx, requestModePut(r), chunk)
 	if err != nil {
-		s.logger.Debugf("chunk upload: chunk write error: %v, addr %s", err, address)
+		s.logger.Debugf("chunk upload: chunk write error: %v, addr %s", err, chunk.Address())
 		s.logger.Error("chunk upload: chunk write error")
 		jsonhttp.BadRequest(w, "chunk write error")
 		return
@@ -123,7 +112,7 @@ func (s *server) chunkUploadHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	w.Header().Set("Access-Control-Expose-Headers", SwarmTagHeader)
-	jsonhttp.OK(w, chunkAddressResponse{Reference: address})
+	jsonhttp.OK(w, chunkAddressResponse{Reference: chunk.Address()})
 }
 
 func (s *server) chunkGetHandler(w http.ResponseWriter, r *http.Request) {

pkg/api/soc.go

@@ -10,16 +10,14 @@ import (
 	"io/ioutil"
 	"net/http"
 
-	"github.com/ethersphere/bee/pkg/bmtpool"
+	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/jsonhttp"
 	"github.com/ethersphere/bee/pkg/soc"
 	"github.com/ethersphere/bee/pkg/swarm"
 	"github.com/gorilla/mux"
 )
 
-var (
-	errBadRequestParams = errors.New("owner, id or span is not well formed")
-)
+var errBadRequestParams = errors.New("owner, id or span is not well formed")
 
 type socPostResponse struct {
 	Reference swarm.Address `json:"reference"`
@@ -82,7 +80,7 @@ func (s *server) socUploadHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	ch, err := chunk(data)
+	ch, err := cac.NewWithDataSpan(data)
 	if err != nil {
 		s.logger.Debugf("soc upload: create content addressed chunk: %v", err)
 		s.logger.Error("soc upload: chunk data error")
@@ -117,18 +115,3 @@ func (s *server) socUploadHandler(w http.ResponseWriter, r *http.Request) {
 
 	jsonhttp.Created(w, chunkAddressResponse{Reference: chunk.Address()})
 }
-
-func chunk(data []byte) (swarm.Chunk, error) {
-	hasher := bmtpool.Get()
-	defer bmtpool.Put(hasher)
-	err := hasher.SetSpanBytes(data[:swarm.SpanSize])
-	if err != nil {
-		return nil, err
-	}
-	_, err = hasher.Write(data[swarm.SpanSize:])
-	if err != nil {
-		return nil, err
-	}
-
-	return swarm.NewChunk(swarm.NewAddress(hasher.Sum(nil)), data), nil
-}

pkg/cac/cac.go

@@ -2,25 +2,72 @@ package cac
 
 import (
 	"encoding/binary"
+	"errors"
 
 	"github.com/ethersphere/bee/pkg/bmtpool"
 	"github.com/ethersphere/bee/pkg/swarm"
 )
 
+var (
+	errTooShortChunkData = errors.New("short chunk data")
+	errTooLargeChunkData = errors.New("data too large")
+)
+
+// New creates a new content address chunk by initializing a span and appending the data to it.
 func New(data []byte) (swarm.Chunk, error) {
-	hasher := bmtpool.Get()
-	defer bmtpool.Put(hasher)
+	dataLength := len(data)
+	if dataLength > swarm.ChunkSize {
+		return nil, errTooLargeChunkData
+	}
 
-	_, err := hasher.Write(data)
-	if err != nil {
-		return nil, err
+	if dataLength == 0 {
+		return nil, errTooShortChunkData
+	}
+
+	span := make([]byte, swarm.SpanSize)
+	binary.LittleEndian.PutUint64(span, uint64(dataLength))
+	return newWithSpan(data, span)
+}
+
+// NewWithDataSpan creates a new chunk assuming that the span precedes the actual data.
+func NewWithDataSpan(data []byte) (swarm.Chunk, error) {
+	dataLength := len(data)
+	if dataLength > swarm.ChunkSize+swarm.SpanSize {
+		return nil, errTooLargeChunkData
 	}
-	span := make([]byte, 8)
-	binary.LittleEndian.PutUint64(span, uint64(len(data)))
-	err = hasher.SetSpanBytes(span)
+
+	if dataLength < swarm.SpanSize {
+		return nil, errTooShortChunkData
+	}
+	return newWithSpan(data[swarm.SpanSize:], data[:swarm.SpanSize])
+}
+
+// newWithSpan creates a new chunk prepending the given span to the data.
+func newWithSpan(data, span []byte) (swarm.Chunk, error) {
+	h := hasher(data)
+	hash, err := h(span)
 	if err != nil {
 		return nil, err
 	}
 
-	return swarm.NewChunk(swarm.NewAddress(hasher.Sum(nil)), append(span, data...)), nil
+	cdata := make([]byte, len(data)+len(span))
+	copy(cdata[:swarm.SpanSize], span)
+	copy(cdata[swarm.SpanSize:], data)
+	return swarm.NewChunk(swarm.NewAddress(hash), cdata), nil
+}
+
+// hasher is a helper function to hash a given data based on the given span.
+func hasher(data []byte) func([]byte) ([]byte, error) {
+	return func(span []byte) ([]byte, error) {
+		hasher := bmtpool.Get()
+		defer bmtpool.Put(hasher)
+
+		if err := hasher.SetSpanBytes(span); err != nil {
+			return nil, err
+		}
+		if _, err := hasher.Write(data); err != nil {
+			return nil, err
+		}
+		return hasher.Sum(nil), nil
+	}
 }

pkg/cac/cac_test.go

 package cac_test
 
 import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/swarm"
 )
 
-func TestCac(t *testing.T) {
-	bmtHashOfFoo := "2387e8e7d8a48c2a9339c97c1dc3461a9a7aa07e994c5cb8b38fd7c1b3e6ea48"
-	address := swarm.MustParseHexAddress(bmtHashOfFoo)
-	foo := "foo"
+func TestNewCAC(t *testing.T) {
+	data := []byte("greaterthanspan")
+	bmtHashOfData := "27913f1bdb6e8e52cbd5a5fd4ab577c857287edf6969b41efe926b51de0f4f23"
+	address := swarm.MustParseHexAddress(bmtHashOfData)
 
-	c, err := cac.New([]byte(foo))
+	expectedSpan := make([]byte, swarm.SpanSize)
+	binary.LittleEndian.PutUint64(expectedSpan, uint64(len(data)))
+	expectedContent := append(expectedSpan, data...)
+
+	c, err := cac.New(data)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !c.Address().Equal(address) {
+		t.Fatalf("address mismatch. got %s want %s", c.Address().String(), address.String())
+	}
+
+	if !bytes.Equal(c.Data(), expectedContent) {
+		t.Fatalf("chunk data mismatch. got %x want %x", c.Data(), expectedContent)
+	}
+}
+
+func TestNewWithDataSpan(t *testing.T) {
+	data := []byte("greaterthanspan")
+	bmtHashOfData := "95022e6af5c6d6a564ee55a67f8455a3e18c511b5697c932d9e44f07f2fb8c53"
+	address := swarm.MustParseHexAddress(bmtHashOfData)
+
+	c, err := cac.NewWithDataSpan(data)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -20,4 +48,56 @@ func TestCac(t *testing.T) {
 	if !c.Address().Equal(address) {
 		t.Fatalf("address mismatch. got %s want %s", c.Address().String(), address.String())
 	}
+
+	if !bytes.Equal(c.Data(), data) {
+		t.Fatalf("chunk data mismatch. got %x want %x", c.Data(), data)
+	}
+}
+
+func TestChunkInvariants(t *testing.T) {
+	chunkerFunc := []struct {
+		name    string
+		chunker func(data []byte) (swarm.Chunk, error)
+	}{
+		{
+			name:    "new cac",
+			chunker: cac.New,
+		},
+		{
+			name:    "new chunk with data span",
+			chunker: cac.NewWithDataSpan,
+		},
+	}
+
+	for _, f := range chunkerFunc {
+		for _, cc := range []struct {
+			name    string
+			data    []byte
+			wantErr error
+		}{
+			{
+				name:    "zero data",
+				data:    []byte{},
+				wantErr: cac.ErrTooShortChunkData,
+			},
+			{
+				name:    "nil",
+				data:    nil,
+				wantErr: cac.ErrTooShortChunkData,
+			},
+			{
+				name:    "too large data chunk",
+				data:    []byte(strings.Repeat("a", swarm.ChunkSize+swarm.SpanSize+1)),
+				wantErr: cac.ErrTooLargeChunkData,
+			},
+		} {
+			testName := fmt.Sprintf("%s-%s", f.name, cc.name)
+			t.Run(testName, func(t *testing.T) {
+				_, err := f.chunker(cc.data)
+				if !errors.Is(err, cc.wantErr) {
+					t.Fatalf("got %v want %v", err, cc.wantErr)
+				}
+			})
+		}
+	}
 }

pkg/cac/export_test.go

+// Copyright 2021 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package cac
+
+var (
+	ErrTooShortChunkData = errTooShortChunkData
+	ErrTooLargeChunkData = errTooLargeChunkData
+)

pkg/feeds/putter.go

@@ -8,7 +8,7 @@ import (
 	"context"
 	"encoding/binary"
 
-	"github.com/ethersphere/bee/pkg/bmtpool"
+	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/crypto"
 	"github.com/ethersphere/bee/pkg/soc"
 	"github.com/ethersphere/bee/pkg/storage"
@@ -57,21 +57,7 @@ func (u *Putter) Put(ctx context.Context, i Index, at int64, payload []byte) err
 }
 
 func toChunk(at uint64, payload []byte) (swarm.Chunk, error) {
-	hasher := bmtpool.Get()
-	defer bmtpool.Put(hasher)
-
 	ts := make([]byte, 8)
 	binary.BigEndian.PutUint64(ts, at)
-	content := append(ts, payload...)
-	_, err := hasher.Write(content)
-	if err != nil {
-		return nil, err
-	}
-	span := make([]byte, 8)
-	binary.LittleEndian.PutUint64(span, uint64(len(content)))
-	err = hasher.SetSpanBytes(span)
-	if err != nil {
-		return nil, err
-	}
-	return swarm.NewChunk(swarm.NewAddress(hasher.Sum(nil)), append(append([]byte{}, span...), content...)), nil
+	return cac.New(append(ts, payload...))
 }

pkg/file/splitter/internal/job.go

@@ -10,7 +10,7 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/ethersphere/bee/pkg/bmtpool"
+	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/encryption"
 	"github.com/ethersphere/bee/pkg/file"
 	"github.com/ethersphere/bee/pkg/sctx"
@@ -136,9 +136,8 @@ func (s *SimpleSplitterJob) sumLevel(lvl int) ([]byte, error) {
 	span := (s.length-1)%spanSize + 1
 
 	var chunkData []byte
-	var addr swarm.Address
 
-	head := make([]byte, 8)
+	head := make([]byte, swarm.SpanSize)
 	binary.LittleEndian.PutUint64(head, uint64(span))
 	tail := s.buffer[s.cursors[lvl+1]:s.cursors[lvl]]
 	chunkData = append(head, tail...)
@@ -157,29 +156,14 @@ func (s *SimpleSplitterJob) sumLevel(lvl int) ([]byte, error) {
 		}
 	}
 
-	hasher := bmtpool.Get()
-
-	err = hasher.SetSpanBytes(c[:8])
-	if err != nil {
-		bmtpool.Put(hasher)
-		return nil, err
-	}
-	_, err = hasher.Write(c[8:])
+	ch, err := cac.NewWithDataSpan(c)
 	if err != nil {
-		bmtpool.Put(hasher)
 		return nil, err
 	}
-	ref := hasher.Sum(nil)
-	bmtpool.Put(hasher)
-
-	addr = swarm.NewAddress(ref)
 
 	// Add tag to the chunk if tag is valid
-	var ch swarm.Chunk
 	if s.tag != nil {
-		ch = swarm.NewChunk(addr, c).WithTagID(s.tag.Uid)
-	} else {
-		ch = swarm.NewChunk(addr, c)
+		ch = ch.WithTagID(s.tag.Uid)
 	}
 
 	seen, err := s.putter.Put(s.ctx, ch)

pkg/soc/export_test.go

@@ -5,7 +5,6 @@
 package soc
 
 var (
-	ToSignDigest          = toSignDigest
-	RecoverAddress        = recoverAddress
-	ContentAddressedChunk = contentAddressedChunk
+	ToSignDigest   = toSignDigest
+	RecoverAddress = recoverAddress
 )

pkg/soc/soc.go

@@ -11,7 +11,7 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/ethersphere/bee/pkg/bmtpool"
+	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/crypto"
 	"github.com/ethersphere/bee/pkg/swarm"
 )
@@ -151,10 +151,7 @@ func FromChunk(sch swarm.Chunk) (*Soc, error) {
 	s.signature = chunkData[cursor : cursor+SignatureSize]
 	cursor += SignatureSize
 
-	spanBytes := chunkData[cursor : cursor+swarm.SpanSize]
-	cursor += swarm.SpanSize
-
-	ch, err := contentAddressedChunk(chunkData[cursor:], spanBytes)
+	ch, err := cac.NewWithDataSpan(chunkData[cursor:])
 	if err != nil {
 		return nil, err
 	}
@@ -258,23 +255,3 @@ func recoverAddress(signature, digest []byte) ([]byte, error) {
 	}
 	return recoveredEthereumAddress, nil
 }
-
-func contentAddressedChunk(data, spanBytes []byte) (swarm.Chunk, error) {
-	hasher := bmtpool.Get()
-	defer bmtpool.Put(hasher)
-
-	// execute hash, compare and return result
-	err := hasher.SetSpanBytes(spanBytes)
-	if err != nil {
-		return nil, err
-	}
-	_, err = hasher.Write(data)
-	if err != nil {
-		return nil, err
-	}
-	s := hasher.Sum(nil)
-
-	payload := append(append([]byte{}, spanBytes...), data...)
-	address := swarm.NewAddress(s)
-	return swarm.NewChunk(address, payload), nil
-}

pkg/soc/soc_test.go

@@ -9,6 +9,7 @@ import (
 	"encoding/binary"
 	"testing"
 
+	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/crypto"
 	"github.com/ethersphere/bee/pkg/soc"
 	"github.com/ethersphere/bee/pkg/swarm"
@@ -26,7 +27,7 @@ func TestToChunk(t *testing.T) {
 	id := make([]byte, 32)
 
 	payload := []byte("foo")
-	ch, err := chunk(payload)
+	ch, err := cac.New(payload)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -93,7 +94,7 @@ func TestFromChunk(t *testing.T) {
 	id := make([]byte, 32)
 
 	payload := []byte("foo")
-	ch, err := chunk(payload)
+	ch, err := cac.New(payload)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -121,9 +122,3 @@ func TestFromChunk(t *testing.T) {
 		t.Fatalf("owner address mismatch %x %x", ownerEthereumAddress, u2.OwnerAddress())
 	}
 }
-
-func chunk(data []byte) (swarm.Chunk, error) {
-	span := make([]byte, swarm.SpanSize)
-	binary.LittleEndian.PutUint64(span, uint64(len(data)))
-	return soc.ContentAddressedChunk(data, span)
-}

pkg/storage/testing/chunk.go

@@ -17,11 +17,10 @@
 package testing
 
 import (
-	"encoding/binary"
 	"math/rand"
 	"time"
 
-	"github.com/ethersphere/bee/pkg/bmtpool"
+	"github.com/ethersphere/bee/pkg/cac"
 	"github.com/ethersphere/bee/pkg/swarm"
 )
 
@@ -55,24 +54,8 @@ func init() {
 func GenerateTestRandomChunk() swarm.Chunk {
 	data := make([]byte, swarm.ChunkSize)
 	_, _ = rand.Read(data)
-	span := make([]byte, swarm.SpanSize)
-	binary.LittleEndian.PutUint64(span, uint64(len(data)))
-	data = append(span, data...)
-
-	hasher := bmtpool.Get()
-	defer bmtpool.Put(hasher)
-
-	err := hasher.SetSpanBytes(data[:swarm.SpanSize])
-	if err != nil {
-		panic(err)
-	}
-	_, err = hasher.Write(data[swarm.SpanSize:])
-	if err != nil {
-		panic(err)
-	}
-	ref := hasher.Sum(nil)
-
-	return swarm.NewChunk(swarm.NewAddress(ref), data)
+	ch, _ := cac.New(data)
+	return ch
 }
 
 // GenerateTestRandomInvalidChunk generates a random, however invalid, content