Merge pull request ethereum#194 from ethersphere/swarm-mutableresource-index

swarm/storage: Improved versioning and reverse lookups for mutable resources

Author: nolash

swarm/storage/resource.go

@@ -29,9 +29,9 @@ type resource struct {
 	name       string
 	ensName    common.Hash
 	startBlock uint64
-	lastBlock  uint64
+	lastPeriod uint32
 	frequency  uint64
-	version    uint64
+	version    uint32
 	data       []byte
 	updated    time.Time
 }
@@ -102,7 +102,7 @@ type ResourceHandler struct {
 	ethapi       *rpc.Client
 	resources    map[string]*resource
 	hashLock     sync.Mutex
-	resourceLock sync.Mutex
+	resourceLock sync.RWMutex
 	hasher       SwarmHash
 	privKey      *ecdsa.PrivateKey
 	maxChunkData int64
@@ -168,7 +168,7 @@ func NewResource(name string, startBlock uint64, frequency uint64) (*resource, e
 	}, nil
 }
 
-// Creates a new root entry for a resource update identified by `name` with the specified `frequency`.
+// Creates a new root entry for a mutable resource identified by `name` with the specified `frequency`.
 //
 // The start block of the resource update will be the actual current block height of the connected network.
 func (self *ResourceHandler) NewResource(name string, frequency uint64) (*resource, error) {
@@ -202,15 +202,15 @@ func (self *ResourceHandler) NewResource(name string, frequency uint64) (*resour
 	self.Put(chunk)
 	log.Debug("new resource", "name", validname, "key", ensName, "startBlock", currentblock, "frequency", frequency)
 
-	self.resourceLock.Lock()
-	defer self.resourceLock.Unlock()
-	self.resources[name] = &resource{
+	rsrc := &resource{
 		name:       validname,
 		ensName:    ensName,
 		startBlock: currentblock,
 		frequency:  frequency,
 		updated:    time.Now(),
 	}
+	self.setResource(name, rsrc)
+
 	return self.resources[name], nil
 }
 
@@ -258,12 +258,12 @@ func (self *ResourceHandler) SetResource(rsrc *resource, allowOverwrite bool) er
 // root chunk.
 // It is the callers responsibility to make sure that this chunk exists (if the resource
 // update root data was retrieved externally, it typically doesn't)
-func (self *ResourceHandler) LookupVersion(name string, nextblock uint64, version uint64, refresh bool) (*resource, error) {
+func (self *ResourceHandler) LookupVersion(name string, period uint32, version uint32, refresh bool) (*resource, error) {
 	rsrc, err := self.loadResource(name, refresh)
 	if err != nil {
 		return nil, err
 	}
-	return self.lookup(rsrc, name, nextblock, version, refresh)
+	return self.lookup(rsrc, name, period, version, refresh)
 }
 
 // Retrieves the latest version of the resource update identified by `name`
@@ -274,12 +274,12 @@ func (self *ResourceHandler) LookupVersion(name string, nextblock uint64, versio
 // and returned.
 //
 // See also (*ResourceHandler).LookupVersion
-func (self *ResourceHandler) LookupHistorical(name string, nextblock uint64, refresh bool) (*resource, error) {
+func (self *ResourceHandler) LookupHistorical(name string, period uint32, refresh bool) (*resource, error) {
 	rsrc, err := self.loadResource(name, refresh)
 	if err != nil {
 		return nil, err
 	}
-	return self.lookup(rsrc, name, nextblock, 0, refresh)
+	return self.lookup(rsrc, name, period, 0, refresh)
 }
 
 // Retrieves the latest version of the resource update identified by `name`
@@ -303,15 +303,15 @@ func (self *ResourceHandler) LookupLatest(name string, refresh bool) (*resource,
 	if err != nil {
 		return nil, err
 	}
-	nextblock := getNextBlock(rsrc.startBlock, currentblock, rsrc.frequency)
-	return self.lookup(rsrc, name, nextblock, 0, refresh)
+	nextperiod := getNextPeriod(rsrc.startBlock, currentblock, rsrc.frequency)
+	return self.lookup(rsrc, name, nextperiod, 0, refresh)
 }
 
 // base code for public lookup methods
-func (self *ResourceHandler) lookup(rsrc *resource, name string, nextblock uint64, version uint64, refresh bool) (*resource, error) {
+func (self *ResourceHandler) lookup(rsrc *resource, name string, period uint32, version uint32, refresh bool) (*resource, error) {
 
-	if nextblock == 0 {
-		return nil, fmt.Errorf("blocknumber must be >0")
+	if period == 0 {
+		return nil, fmt.Errorf("period must be >0")
 	}
 
 	// start from the last possible block period, and iterate previous ones until we find a match
@@ -323,29 +323,29 @@ func (self *ResourceHandler) lookup(rsrc *resource, name string, nextblock uint6
 		version = 1
 	}
 
-	for nextblock > rsrc.startBlock {
-		key := self.resourceHash(rsrc.ensName, nextblock, version)
+	for period > 0 {
+		key := self.resourceHash(rsrc.ensName, period, version)
 		chunk, err := self.Get(key)
 		if err == nil {
 			if specificversion {
-				return self.updateResourceIndex(rsrc, chunk, nextblock, version, &name)
+				return self.updateResourceIndex(rsrc, chunk, &name)
 			}
 			// check if we have versions > 1. If a version fails, the previous version is used and returned.
-			log.Trace("rsrc update version 1 found, checking for version updates", "nextblock", nextblock, "key", key)
+			log.Trace("rsrc update version 1 found, checking for version updates", "period", period, "key", key)
 			for {
 				newversion := version + 1
-				key := self.resourceHash(rsrc.ensName, nextblock, newversion)
+				key := self.resourceHash(rsrc.ensName, period, newversion)
 				newchunk, err := self.Get(key)
 				if err != nil {
-					return self.updateResourceIndex(rsrc, chunk, nextblock, version, &name)
+					return self.updateResourceIndex(rsrc, chunk, &name)
 				}
-				log.Trace("version update found, checking next", "version", version, "block", nextblock, "key", key)
+				log.Trace("version update found, checking next", "version", version, "period", period, "key", key)
 				chunk = newchunk
 				version = newversion
 			}
 		}
-		log.Trace("rsrc update not found, checking previous period", "block", nextblock, "key", key)
-		nextblock -= rsrc.frequency
+		log.Trace("rsrc update not found, checking previous period", "period", period, "key", key)
+		period--
 	}
 	return nil, fmt.Errorf("no updates found")
 }
@@ -355,12 +355,9 @@ func (self *ResourceHandler) loadResource(name string, refresh bool) (*resource,
 	// if the resource is not known to this session we must load it
 	// if refresh is set, we force load
 
-	rsrc := &resource{}
-
-	self.resourceLock.Lock()
-	_, ok := self.resources[name]
-	self.resourceLock.Unlock()
-	if !ok || refresh {
+	rsrc := self.getResource(name)
+	if rsrc == nil || refresh {
+		rsrc = &resource{}
 		// make sure our ens identifier is idna safe
 		validname, err := idna.ToASCII(name)
 		if err != nil {
@@ -397,7 +394,7 @@ func (self *ResourceHandler) loadResource(name string, refresh bool) (*resource,
 }
 
 // update mutable resource index map with specified content
-func (self *ResourceHandler) updateResourceIndex(rsrc *resource, chunk *Chunk, nextblock uint64, version uint64, indexname *string) (*resource, error) {
+func (self *ResourceHandler) updateResourceIndex(rsrc *resource, chunk *Chunk, indexname *string) (*resource, error) {
 
 	// rsrc update data chunks are total hacks
 	// and have no size prefix :D
@@ -407,18 +404,36 @@ func (self *ResourceHandler) updateResourceIndex(rsrc *resource, chunk *Chunk, n
 	}
 
 	// update our rsrcs entry map
-	rsrc.lastBlock = nextblock
+	period, version, _, data, err := parseUpdate(chunk.SData[signatureLength:])
+	rsrc.lastPeriod = period
 	rsrc.version = version
-	rsrc.data = make([]byte, len(chunk.SData)-signatureLength)
 	rsrc.updated = time.Now()
-	copy(rsrc.data, chunk.SData[signatureLength:])
-	log.Debug("Resource synced", "name", rsrc.name, "key", chunk.Key, "block", nextblock, "version", version)
-	self.resourceLock.Lock()
-	self.resources[*indexname] = rsrc
-	self.resourceLock.Unlock()
+	rsrc.data = make([]byte, len(data))
+	copy(rsrc.data, data)
+	log.Debug("Resource synced", "name", rsrc.name, "key", chunk.Key, "period", rsrc.lastPeriod, "version", rsrc.version)
+	self.setResource(*indexname, rsrc)
 	return rsrc, nil
 }
 
+func parseUpdate(blob []byte) (period uint32, version uint32, ensname []byte, data []byte, err error) {
+	headerlength := binary.LittleEndian.Uint16(blob[:2])
+	if int(headerlength+2) > len(blob) {
+		return 0, 0, nil, nil, fmt.Errorf("Reported header length %d longer than actual data length %d", headerlength, len(blob))
+	}
+	cursor := 2
+	period = binary.LittleEndian.Uint32(blob[cursor : cursor+4])
+	cursor += 4
+	version = binary.LittleEndian.Uint32(blob[cursor : cursor+4])
+	cursor += 4
+	namelength := int(headerlength) - cursor + 2
+	ensname = make([]byte, namelength)
+	copy(ensname, blob[cursor:])
+	cursor += namelength
+	data = make([]byte, len(blob)-cursor)
+	copy(data, blob[cursor:])
+	return
+}
+
 // Adds an actual data update
 //
 // Uses the data currently loaded in the resources map entry.
@@ -447,28 +462,48 @@ func (self *ResourceHandler) Update(name string, data []byte) (Key, error) {
 	if err != nil {
 		return nil, err
 	}
-	nextblock := getNextBlock(resource.startBlock, currentblock, resource.frequency)
+	nextperiod := getNextPeriod(resource.startBlock, currentblock, resource.frequency)
 
 	// if we already have an update for this block then increment version
-	var version uint64
-	if nextblock == resource.lastBlock {
+	var version uint32
+	if self.hasUpdate(name, nextperiod) {
 		version = resource.version
 	}
 	version++
 
+	// prepend version and period to allow reverse lookups
+	// data header length does NOT include the header length prefix bytes themselves
+	headerlength := uint16(len(resource.ensName) + 4 + 4)
+	fulldata := make([]byte, int(headerlength)+2+len(data))
+
+	cursor := 0
+	binary.LittleEndian.PutUint16(fulldata, headerlength)
+	cursor += 2
+
+	binary.LittleEndian.PutUint32(fulldata[cursor:], nextperiod)
+	cursor += 4
+
+	binary.LittleEndian.PutUint32(fulldata[cursor:], version)
+	cursor += 4
+
+	copy(fulldata[cursor:], resource.ensName[:])
+	cursor += len(resource.ensName)
+
+	copy(fulldata[cursor:], data)
+
 	// create the update chunk and send it
-	key := self.resourceHash(resource.ensName, nextblock, version)
+	key := self.resourceHash(resource.ensName, nextperiod, version)
 	chunk := NewChunk(key, nil)
-	chunk.SData, err = self.signContent(data)
+	chunk.SData, err = self.signContent(fulldata)
 	if err != nil {
 		return nil, err
 	}
-	chunk.Size = int64(len(data))
+	chunk.Size = int64(len(fulldata))
 	self.Put(chunk)
-	log.Trace("resource update", "name", resource.name, "key", key, "currentblock", currentblock, "lastBlock", nextblock, "version", version)
+	log.Trace("resource update", "name", resource.name, "key", key, "currentblock", currentblock, "lastperiod", nextperiod, "version", version, "data", chunk.SData)
 
 	// update our resources map entry and return the new key
-	resource.lastBlock = nextblock
+	resource.lastPeriod = nextperiod
 	resource.version = version
 	resource.data = make([]byte, len(data))
 	copy(resource.data, data)
@@ -494,20 +529,37 @@ func (self *ResourceHandler) getBlock() (uint64, error) {
 	return strconv.ParseUint(currentblock, 10, 64)
 }
 
-func (self *ResourceHandler) resourceHash(namehash common.Hash, blockheight uint64, version uint64) Key {
-	// format is: hash(namehash|blockheight|version)
+func (self *ResourceHandler) BlockToPeriod(name string, blocknumber uint64) uint32 {
+	return getNextPeriod(self.resources[name].startBlock, blocknumber, self.resources[name].frequency)
+}
+
+func (self *ResourceHandler) PeriodToBlock(name string, period uint32) uint64 {
+	return self.resources[name].startBlock + (uint64(period) * self.resources[name].frequency)
+}
+
+func (self *ResourceHandler) getResource(name string) *resource {
+	self.resourceLock.RLock()
+	defer self.resourceLock.RUnlock()
+	rsrc := self.resources[name]
+	return rsrc
+}
+
+func (self *ResourceHandler) setResource(name string, rsrc *resource) {
+	self.resourceLock.Lock()
+	defer self.resourceLock.Unlock()
+	self.resources[name] = rsrc
+}
+
+func (self *ResourceHandler) resourceHash(namehash common.Hash, period uint32, version uint32) Key {
+	// format is: hash(namehash|period|version)
 	self.hashLock.Lock()
 	defer self.hashLock.Unlock()
 	self.hasher.Reset()
 	self.hasher.Write(namehash[:])
-	b := make([]byte, 8)
-	c := binary.PutUvarint(b, blockheight)
+	b := make([]byte, 4)
+	binary.LittleEndian.PutUint32(b, period)
 	self.hasher.Write(b)
-	// PutUvarint only overwrites first c bytes
-	for i := 0; i < c; i++ {
-		b[i] = 0
-	}
-	c = binary.PutUvarint(b, version)
+	binary.LittleEndian.PutUint32(b, version)
 	self.hasher.Write(b)
 	return self.hasher.Sum(nil)
 }
@@ -554,6 +606,13 @@ func (self *ResourceHandler) verifyContent(chunkdata []byte) error {
 	return nil
 }
 
+func (self *ResourceHandler) hasUpdate(name string, period uint32) bool {
+	if self.resources[name].lastPeriod == period {
+		return true
+	}
+	return false
+}
+
 type resourceChunkStore struct {
 	localStore ChunkStore
 	netStore   ChunkStore
@@ -596,8 +655,8 @@ func (r *resourceChunkStore) Close() {
 	r.localStore.Close()
 }
 
-func getNextBlock(start uint64, current uint64, frequency uint64) uint64 {
+func getNextPeriod(start uint64, current uint64, frequency uint64) uint32 {
 	blockdiff := current - start
-	periods := (blockdiff / frequency) + 1
-	return start + (frequency * periods)
+	period := blockdiff / frequency
+	return uint32(period + 1)
 }