Merge branch 'swarm-network-rewrite' into kademlia-fixes

swarm/api/api.go

@@ -39,7 +39,8 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
-var hashMatcher = regexp.MustCompile("^[0-9A-Fa-f]{64}")
+// TODO: this is bad, it should not be hardcoded how long is a hash
+var hashMatcher = regexp.MustCompile("^([0-9A-Fa-f]{64})([0-9A-Fa-f]{64})?")
 
 type ErrResourceReturn struct {
 	key string
@@ -230,9 +231,9 @@ func NewApi(dpa *storage.DPA, dns Resolver, resourceHandler *storage.ResourceHan
 }
 
 // to be used only in TEST
-func (self *Api) Upload(uploadDir, index string) (hash string, err error) {
+func (self *Api) Upload(uploadDir, index string, toEncrypt bool) (hash string, err error) {
 	fs := NewFileSystem(self)
-	hash, err = fs.Upload(uploadDir, index)
+	hash, err = fs.Upload(uploadDir, index, toEncrypt)
 	return hash, err
 }
 
@@ -241,9 +242,9 @@ func (self *Api) Retrieve(key storage.Key) storage.LazySectionReader {
 	return self.dpa.Retrieve(key)
 }
 
-func (self *Api) Store(data io.Reader, size int64) (key storage.Key, wait func(), err error) {
+func (self *Api) Store(data io.Reader, size int64, toEncrypt bool) (key storage.Key, wait func(), err error) {
 	log.Debug("api.store", "size", size)
-	return self.dpa.Store(data, size, false)
+	return self.dpa.Store(data, size, toEncrypt)
 }
 
 type ErrResolve error
@@ -283,17 +284,17 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
 }
 
 // Put provides singleton manifest creation on top of dpa store
-func (self *Api) Put(content, contentType string) (k storage.Key, wait func(), err error) {
+func (self *Api) Put(content, contentType string, toEncrypt bool) (k storage.Key, wait func(), err error) {
 	apiPutCount.Inc(1)
 	r := strings.NewReader(content)
-	key, waitContent, err := self.dpa.Store(r, int64(len(content)), false)
+	key, waitContent, err := self.dpa.Store(r, int64(len(content)), toEncrypt)
 	if err != nil {
 		apiPutFail.Inc(1)
 		return nil, nil, err
 	}
 	manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType)
 	r = strings.NewReader(manifest)
-	key, waitManifest, err := self.dpa.Store(r, int64(len(manifest)), false)
+	key, waitManifest, err := self.dpa.Store(r, int64(len(manifest)), toEncrypt)
 	if err != nil {
 		apiPutFail.Inc(1)
 		return nil, nil, err
@@ -377,7 +378,7 @@ func (self *Api) Modify(key storage.Key, path, contentHash, contentType string)
 		apiModifyFail.Inc(1)
 		return nil, err
 	}
-	return trie.hash, nil
+	return trie.ref, nil
 }
 
 func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver bool) (storage.Key, string, error) {

swarm/api/api_test.go

@@ -32,7 +32,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
-func testApi(t *testing.T, f func(*Api)) {
+func testApi(t *testing.T, f func(*Api, bool)) {
 	datadir, err := ioutil.TempDir("", "bzz-test")
 	if err != nil {
 		t.Fatalf("unable to create temp dir: %v", err)
@@ -43,7 +43,8 @@ func testApi(t *testing.T, f func(*Api)) {
 		return
 	}
 	api := NewApi(dpa, nil, nil)
-	f(api)
+	f(api, false)
+	f(api, true)
 }
 
 type testResponse struct {
@@ -106,11 +107,11 @@ func testGet(t *testing.T, api *Api, bzzhash, path string) *testResponse {
 }
 
 func TestApiPut(t *testing.T) {
-	testApi(t, func(api *Api) {
+	testApi(t, func(api *Api, toEncrypt bool) {
 		content := "hello"
 		exp := expResponse(content, "text/plain", 0)
 		// exp := expResponse([]byte(content), "text/plain", 0)
-		key, wait, err := api.Put(content, exp.MimeType)
+		key, wait, err := api.Put(content, exp.MimeType, toEncrypt)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}

swarm/api/filesystem.go

@@ -47,7 +47,7 @@ func NewFileSystem(api *Api) *FileSystem {
 // TODO: localpath should point to a manifest
 //
 // DEPRECATED: Use the HTTP API instead
-func (self *FileSystem) Upload(lpath, index string) (string, error) {
+func (self *FileSystem) Upload(lpath, index string, toEncrypt bool) (string, error) {
 	var list []*manifestTrieEntry
 	localpath, err := filepath.Abs(filepath.Clean(lpath))
 	if err != nil {
@@ -114,7 +114,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
 				stat, _ := f.Stat()
 				var hash storage.Key
 				var wait func()
-				hash, wait, err = self.api.dpa.Store(f, stat.Size(), false)
+				hash, wait, err = self.api.dpa.Store(f, stat.Size(), toEncrypt)
 				if hash != nil {
 					list[i].Hash = hash.Hex()
 				}
@@ -164,7 +164,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
 	err2 := trie.recalcAndStore()
 	var hs string
 	if err2 == nil {
-		hs = trie.hash.Hex()
+		hs = trie.ref.Hex()
 	}
 	awg.Wait()
 	return hs, err2

swarm/api/filesystem_test.go

@@ -29,9 +29,9 @@ import (
 
 var testDownloadDir, _ = ioutil.TempDir(os.TempDir(), "bzz-test")
 
-func testFileSystem(t *testing.T, f func(*FileSystem)) {
+func testFileSystem(t *testing.T, f func(*FileSystem, bool)) {
-	testApi(t, func(api *Api) {
+	testApi(t, func(api *Api, toEncrypt bool) {
-		f(NewFileSystem(api))
+		f(NewFileSystem(api), toEncrypt)
 	})
 }
 
@@ -46,9 +46,9 @@ func readPath(t *testing.T, parts ...string) string {
 }
 
 func TestApiDirUpload0(t *testing.T) {
-	testFileSystem(t, func(fs *FileSystem) {
+	testFileSystem(t, func(fs *FileSystem, toEncrypt bool) {
 		api := fs.api
-		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0"), "")
+		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0"), "", toEncrypt)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
@@ -74,20 +74,21 @@ func TestApiDirUpload0(t *testing.T) {
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
-		newbzzhash, err := fs.Upload(downloadDir, "")
+		newbzzhash, err := fs.Upload(downloadDir, "", toEncrypt)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
-		if bzzhash != newbzzhash {
+		// TODO: currently the hash is not deterministic in the encrypted case
+		if !toEncrypt && bzzhash != newbzzhash {
 			t.Fatalf("download %v reuploaded has incorrect hash, expected %v, got %v", downloadDir, bzzhash, newbzzhash)
 		}
 	})
 }
 
 func TestApiDirUploadModify(t *testing.T) {
-	testFileSystem(t, func(fs *FileSystem) {
+	testFileSystem(t, func(fs *FileSystem, toEncrypt bool) {
 		api := fs.api
-		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0"), "")
+		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0"), "", toEncrypt)
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
 			return
@@ -104,7 +105,7 @@ func TestApiDirUploadModify(t *testing.T) {
 			t.Errorf("unexpected error: %v", err)
 			return
 		}
-		hash, wait, err := api.Store(bytes.NewReader(index), int64(len(index)))
+		hash, wait, err := api.Store(bytes.NewReader(index), int64(len(index)), toEncrypt)
 		wait()
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
@@ -144,9 +145,9 @@ func TestApiDirUploadModify(t *testing.T) {
 }
 
 func TestApiDirUploadWithRootFile(t *testing.T) {
-	testFileSystem(t, func(fs *FileSystem) {
+	testFileSystem(t, func(fs *FileSystem, toEncrypt bool) {
 		api := fs.api
-		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0"), "index.html")
+		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0"), "index.html", toEncrypt)
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
 			return
@@ -160,9 +161,9 @@ func TestApiDirUploadWithRootFile(t *testing.T) {
 }
 
 func TestApiFileUpload(t *testing.T) {
-	testFileSystem(t, func(fs *FileSystem) {
+	testFileSystem(t, func(fs *FileSystem, toEncrypt bool) {
 		api := fs.api
-		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0", "index.html"), "")
+		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0", "index.html"), "", toEncrypt)
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
 			return
@@ -176,9 +177,9 @@ func TestApiFileUpload(t *testing.T) {
 }
 
 func TestApiFileUploadWithRootFile(t *testing.T) {
-	testFileSystem(t, func(fs *FileSystem) {
+	testFileSystem(t, func(fs *FileSystem, toEncrypt bool) {
 		api := fs.api
-		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0", "index.html"), "index.html")
+		bzzhash, err := fs.Upload(filepath.Join("testdata", "test0", "index.html"), "index.html", toEncrypt)
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
 			return

swarm/api/http/server.go

@@ -124,19 +124,30 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
 	log.Debug("handle.post.raw", "ruid", r.ruid)
 
 	postRawCount.Inc(1)
+
+	toEncrypt := false
+	if r.uri.Addr == "encrypt" {
+		toEncrypt = true
+	}
+
 	if r.uri.Path != "" {
 		postRawFail.Inc(1)
 		Respond(w, r, "raw POST request cannot contain a path", http.StatusBadRequest)
 		return
 	}
 
+	if r.uri.Addr != "" && r.uri.Addr != "encrypt" {
+		postRawFail.Inc(1)
+		Respond(w, r, "raw POST request addr can only be empty or \"encrypt\"", http.StatusBadRequest)
+		return
+	}
+
 	if r.Header.Get("Content-Length") == "" {
 		postRawFail.Inc(1)
 		Respond(w, r, "missing Content-Length header in request", http.StatusBadRequest)
 		return
 	}
-
+	key, _, err := s.api.Store(r.Body, r.ContentLength, toEncrypt)
-	key, _, err := s.api.Store(r.Body, r.ContentLength)
 	if err != nil {
 		postRawFail.Inc(1)
 		Respond(w, r, err.Error(), http.StatusInternalServerError)
@@ -166,8 +177,13 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 		return
 	}
 
+	toEncrypt := false
+	if r.uri.Addr == "encrypt" {
+		toEncrypt = true
+	}
+
 	var key storage.Key
-	if r.uri.Addr != "" {
+	if r.uri.Addr != "" && r.uri.Addr != "encrypt" {
 		key, err = s.api.Resolve(r.uri)
 		if err != nil {
 			postFilesFail.Inc(1)
@@ -176,7 +192,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 		}
 		log.Debug("resolved key", "ruid", r.ruid, "key", key)
 	} else {
-		key, err = s.api.NewManifest()
+		key, err = s.api.NewManifest(toEncrypt)
 		if err != nil {
 			postFilesFail.Inc(1)
 			Respond(w, r, err.Error(), http.StatusInternalServerError)
@@ -840,14 +856,18 @@ func (s *Server) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 	req.uri = uri
 
 	log.Debug("parsed request path", "ruid", req.ruid, "method", req.Method, "uri", req.uri)
+	log.Debug("parsed request path", "uri.Addr", req.uri.Addr, "uri.path", req.uri.Path, "uri.Scheme", req.uri.Scheme)
 
 	switch r.Method {
 	case "POST":
 		if uri.Raw() || uri.DeprecatedRaw() {
+			log.Debug("handlePostRaw")
 			s.HandlePostRaw(w, req)
 		} else if uri.Resource() {
+			log.Debug("handlePostResource")
 			s.HandlePostResource(w, req)
 		} else {
+			log.Debug("handlePostFiles")
 			s.HandlePostFiles(w, req)
 		}
 

swarm/api/manifest.go

@@ -59,13 +59,13 @@ type ManifestList struct {
 }
 
 // NewManifest creates and stores a new, empty manifest
-func (a *Api) NewManifest() (storage.Key, error) {
+func (a *Api) NewManifest(toEncrypt bool) (storage.Key, error) {
 	var manifest Manifest
 	data, err := json.Marshal(&manifest)
 	if err != nil {
 		return nil, err
 	}
-	key, wait, err := a.Store(bytes.NewReader(data), int64(len(data)))
+	key, wait, err := a.Store(bytes.NewReader(data), int64(len(data)), toEncrypt)
 	wait()
 	return key, err
 }
@@ -83,7 +83,7 @@ func (a *Api) NewResourceManifest(resourceKey string) (storage.Key, error) {
 	if err != nil {
 		return nil, err
 	}
-	key, _, err := a.Store(bytes.NewReader(data), int64(len(data)))
+	key, _, err := a.Store(bytes.NewReader(data), int64(len(data)), false)
 	return key, err
 }
 
@@ -104,7 +104,8 @@ func (a *Api) NewManifestWriter(key storage.Key, quitC chan bool) (*ManifestWrit
 
 // AddEntry stores the given data and adds the resulting key to the manifest
 func (m *ManifestWriter) AddEntry(data io.Reader, e *ManifestEntry) (storage.Key, error) {
-	key, _, err := m.api.Store(data, e.Size)
+
+	key, _, err := m.api.Store(data, e.Size, m.trie.encrypted)
 	if err != nil {
 		return nil, err
 	}
@@ -122,7 +123,7 @@ func (m *ManifestWriter) RemoveEntry(path string) error {
 
 // Store stores the manifest, returning the resulting storage key
 func (m *ManifestWriter) Store() (storage.Key, error) {
-	return m.trie.hash, m.trie.recalcAndStore()
+	return m.trie.ref, m.trie.recalcAndStore()
 }
 
 // ManifestWalker is used to recursively walk the entries in the manifest and
@@ -182,9 +183,10 @@ func (m *ManifestWalker) walk(trie *manifestTrie, prefix string, walkFn WalkFn)
 }
 
 type manifestTrie struct {
-	dpa     *storage.DPA
+	dpa       *storage.DPA
-	entries [257]*manifestTrieEntry // indexed by first character of basePath, entries[256] is the empty basePath entry
+	entries   [257]*manifestTrieEntry // indexed by first character of basePath, entries[256] is the empty basePath entry
-	hash    storage.Key             // if hash != nil, it is stored
+	ref       storage.Key             // if ref != nil, it is stored
+	encrypted bool
 }
 
 func newManifestTrieEntry(entry *ManifestEntry, subtrie *manifestTrie) *manifestTrieEntry {
@@ -228,7 +230,7 @@ func readManifest(manifestReader storage.LazySectionReader, hash storage.Key, dp
 		return
 	}
 
-	log.Trace("manifest retrieved", "key", hash)
+	log.Debug("manifest retrieved", "key", hash)
 	var man struct {
 		Entries []*manifestTrieEntry `json:"entries"`
 	}
@@ -242,7 +244,8 @@ func readManifest(manifestReader storage.LazySectionReader, hash storage.Key, dp
 	log.Trace("manifest entries", "key", hash, "len", len(man.Entries))
 
 	trie = &manifestTrie{
-		dpa: dpa,
+		dpa:       dpa,
+		encrypted: (len(hash) > dpa.HashSize()),
 	}
 	for _, entry := range man.Entries {
 		trie.addEntry(entry, quitC)
@@ -251,7 +254,7 @@ func readManifest(manifestReader storage.LazySectionReader, hash storage.Key, dp
 }
 
 func (self *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) {
-	self.hash = nil // trie modified, hash needs to be re-calculated on demand
+	self.ref = nil // trie modified, hash needs to be re-calculated on demand
 
 	if len(entry.Path) == 0 {
 		self.entries[256] = entry
@@ -283,7 +286,8 @@ func (self *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) {
 	commonPrefix := entry.Path[:cpl]
 
 	subtrie := &manifestTrie{
-		dpa: self.dpa,
+		dpa:       self.dpa,
+		encrypted: self.encrypted,
 	}
 	entry.Path = entry.Path[cpl:]
 	oldentry.Path = oldentry.Path[cpl:]
@@ -307,7 +311,7 @@ func (self *manifestTrie) getCountLast() (cnt int, entry *manifestTrieEntry) {
 }
 
 func (self *manifestTrie) deleteEntry(path string, quitC chan bool) {
-	self.hash = nil // trie modified, hash needs to be re-calculated on demand
+	self.ref = nil // trie modified, hash needs to be re-calculated on demand
 
 	if len(path) == 0 {
 		self.entries[256] = nil
@@ -343,7 +347,7 @@ func (self *manifestTrie) deleteEntry(path string, quitC chan bool) {
 }
 
 func (self *manifestTrie) recalcAndStore() error {
-	if self.hash != nil {
+	if self.ref != nil {
 		return nil
 	}
 
@@ -358,7 +362,7 @@ func (self *manifestTrie) recalcAndStore() error {
 				if err != nil {
 					return err
 				}
-				entry.Hash = entry.subtrie.hash.Hex()
+				entry.Hash = entry.subtrie.ref.Hex()
 			}
 			list.Entries = append(list.Entries, entry.ManifestEntry)
 		}
@@ -371,9 +375,9 @@ func (self *manifestTrie) recalcAndStore() error {
 	}
 
 	sr := bytes.NewReader(manifest)
-	key, wait, err2 := self.dpa.Store(sr, int64(len(manifest)), false)
+	key, wait, err2 := self.dpa.Store(sr, int64(len(manifest)), self.encrypted)
 	wait()
-	self.hash = key
+	self.ref = key
 	return err2
 }
 

swarm/api/manifest_test.go

@@ -42,7 +42,9 @@ func manifest(paths ...string) (manifestReader storage.LazySectionReader) {
 
 func testGetEntry(t *testing.T, path, match string, multiple bool, paths ...string) *manifestTrie {
 	quitC := make(chan bool)
-	trie, err := readManifest(manifest(paths...), nil, nil, quitC)
+	dpa := storage.NewDPA(nil, storage.NewDPAParams())
+	ref := make([]byte, dpa.HashSize())
+	trie, err := readManifest(manifest(paths...), ref, dpa, quitC)
 	if err != nil {
 		t.Errorf("unexpected error making manifest: %v", err)
 	}
@@ -97,7 +99,9 @@ func TestGetEntry(t *testing.T) {
 func TestExactMatch(t *testing.T) {
 	quitC := make(chan bool)
 	mf := manifest("shouldBeExactMatch.css", "shouldBeExactMatch.css.map")
-	trie, err := readManifest(mf, nil, nil, quitC)
+	dpa := storage.NewDPA(nil, storage.NewDPAParams())
+	ref := make([]byte, dpa.HashSize())
+	trie, err := readManifest(mf, ref, dpa, quitC)
 	if err != nil {
 		t.Errorf("unexpected error making manifest: %v", err)
 	}
@@ -128,7 +132,9 @@ func TestAddFileWithManifestPath(t *testing.T) {
 	reader := &storage.LazyTestSectionReader{
 		SectionReader: io.NewSectionReader(bytes.NewReader(manifest), 0, int64(len(manifest))),
 	}
-	trie, err := readManifest(reader, nil, nil, nil)
+	dpa := storage.NewDPA(nil, storage.NewDPAParams())
+	ref := make([]byte, dpa.HashSize())
+	trie, err := readManifest(reader, ref, dpa, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

swarm/api/storage.go

@@ -45,8 +45,8 @@ func NewStorage(api *Api) *Storage {
 // its content type
 //
 // DEPRECATED: Use the HTTP API instead
-func (self *Storage) Put(content, contentType string) (storage.Key, func(), error) {
+func (self *Storage) Put(content, contentType string, toEncrypt bool) (storage.Key, func(), error) {
-	return self.api.Put(content, contentType)
+	return self.api.Put(content, contentType, toEncrypt)
 }
 
 // Get retrieves the content from bzzpath and reads the response in full

swarm/api/storage_test.go

@@ -20,18 +20,18 @@ import (
 	"testing"
 )
 
-func testStorage(t *testing.T, f func(*Storage)) {
+func testStorage(t *testing.T, f func(*Storage, bool)) {
-	testApi(t, func(api *Api) {
+	testApi(t, func(api *Api, toEncrypt bool) {
-		f(NewStorage(api))
+		f(NewStorage(api), toEncrypt)
 	})
 }
 
 func TestStoragePutGet(t *testing.T) {
-	testStorage(t, func(api *Storage) {
+	testStorage(t, func(api *Storage, toEncrypt bool) {
 		content := "hello"
 		exp := expResponse(content, "text/plain", 0)
 		// exp := expResponse([]byte(content), "text/plain", 0)
-		bzzkey, wait, err := api.Put(content, exp.MimeType)
+		bzzkey, wait, err := api.Put(content, exp.MimeType, toEncrypt)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}

swarm/pss/pss.go

@@ -177,7 +177,10 @@ func (self *Pss) Start(srv *p2p.Server) error {
 	go func() {
 		for {
 			tickC := time.Tick(defaultCleanInterval)
+			cacheTickC := time.Tick(self.cacheTTL)
 			select {
+			case <-cacheTickC:
+				self.cleanFwdCache()
 			case <-tickC:
 				self.cleanKeys()
 			case <-self.quitC:
@@ -756,6 +759,17 @@ func (self *Pss) forward(msg *PssMsg) {
 // SECTION: Caching
 /////////////////////////////////////////////////////////////////////
 
+// cleanFwdCache is used to periodically remove expired entries from the forward cache
+func (self *Pss) cleanFwdCache() {
+	self.fwdCacheMu.Lock()
+	defer self.fwdCacheMu.Unlock()
+	for k, v := range self.fwdCache {
+		if v.expiresAt.Before(time.Now()) {
+			delete(self.fwdCache, k)
+		}
+	}
+}
+
 // add a message to the cache
 func (self *Pss) addFwdCache(msg *PssMsg) error {
 	var entry pssCacheEntry

swarm/pss/pss_test.go

@@ -147,6 +147,7 @@ func TestCache(t *testing.T) {
 	pp := NewPssParams(privkey)
 	data := []byte("foo")
 	datatwo := []byte("bar")
+	datathree := []byte("baz")
 	wparams := &whisper.MessageParams{
 		TTL:      defaultWhisperTTL,
 		Src:      privkey,
@@ -169,6 +170,13 @@ func TestCache(t *testing.T) {
 		Payload: envtwo,
 		To:      to,
 	}
+	wparams.Payload = datathree
+	woutmsg, err = whisper.NewSentMessage(wparams)
+	envthree, err := woutmsg.Wrap(wparams)
+	msgthree := &PssMsg{
+		Payload: envthree,
+		To:      to,
+	}
 
 	digest := ps.digest(msg)
 	if err != nil {
@@ -178,6 +186,10 @@ func TestCache(t *testing.T) {
 	if err != nil {
 		t.Fatalf("could not store cache msgtwo: %v", err)
 	}
+	digestthree := ps.digest(msgthree)
+	if err != nil {
+		t.Fatalf("could not store cache msgthree: %v", err)
+	}
 
 	if digest == digesttwo {
 		t.Fatalf("different msgs return same hash: %d", digesttwo)
@@ -197,10 +209,23 @@ func TestCache(t *testing.T) {
 		t.Fatalf("message %v should NOT have EXPIRE record in cache but checkCache returned true", msgtwo)
 	}
 
-	time.Sleep(pp.CacheTTL)
+	time.Sleep(pp.CacheTTL + 1*time.Second)
+	err = ps.addFwdCache(msgthree)
+	if err != nil {
+		t.Fatalf("write to pss expire cache failed: %v", err)
+	}
+
 	if ps.checkFwdCache(msg) {
 		t.Fatalf("message %v should have expired from cache but checkCache returned true", msg)
 	}
+
+	if _, ok := ps.fwdCache[digestthree]; !ok {
+		t.Fatalf("unexpired message should be in the cache: %v", digestthree)
+	}
+
+	if _, ok := ps.fwdCache[digesttwo]; ok {
+		t.Fatalf("expired message should have been cleared from the cache: %v", digesttwo)
+	}
 }
 
 // matching of address hints; whether a message could be or is for the node
@@ -1309,6 +1334,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss
 		pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity
 	}
 	ps := NewPss(overlay, pp)
+	ps.Start(nil)
 
 	return ps
 }

swarm/storage/dpa.go

@@ -100,3 +100,7 @@ func (self *DPA) Store(data io.Reader, size int64, toEncrypt bool) (key Key, wai
 	putter := NewHasherStore(self.ChunkStore, self.hashFunc, toEncrypt)
 	return PyramidSplit(data, putter, putter)
 }
+
+func (self *DPA) HashSize() int {
+	return self.hashFunc().Size()
+}