Fix panic in hinted handoff processor

A short write has occurred and we do not have enough bytes to determine
the size of the payload.  This is corrupted record that we should drop.
Instead of panicing, log the error and advance the queue since the error
at this location is unreoverable currently.

Fixes #3436

CHANGELOG.md

@@ -24,6 +24,7 @@
 - [#3579](https://github.com/influxdb/influxdb/issues/3579): Revert breaking change to `client.NewClient` function
 - [#3580](https://github.com/influxdb/influxdb/issues/3580): Do not allow wildcards with fields in select statements
 - [#3530](https://github.com/influxdb/influxdb/pull/3530): Aliasing a column no longer works
+- [#3436](https://github.com/influxdb/influxdb/issues/3436): Fix panic in hinted handoff queue processor
 
 ## v0.9.2 [2015-07-24]
 

services/hh/processor.go

@@ -145,10 +145,10 @@ func (p *Processor) Process() error {
 				// unmarshal the byte slice back to shard ID and points
 				shardID, points, err := p.unmarshalWrite(buf)
 				if err != nil {
-					// TODO: If we ever get and error here, we should probably drop the
-					// the write and let anti-entropy resolve it.  This would be an urecoverable
-					// error and could block the queue indefinitely.
-					res <- err
+					p.Logger.Printf("unmarshal write failed: %v", err)
+					if err := q.Advance(); err != nil {
+						res <- err
+					}
 					return
 				}
 
@@ -197,6 +197,9 @@ func (p *Processor) marshalWrite(shardID uint64, points []tsdb.Point) []byte {
 }
 
 func (p *Processor) unmarshalWrite(b []byte) (uint64, []tsdb.Point, error) {
+	if len(b) < 8 {
+		return 0, nil, fmt.Errorf("too short: len = %d", len(b))
+	}
 	ownerID := binary.BigEndian.Uint64(b[:8])
 	points, err := tsdb.ParsePoints(b[8:])
 	return ownerID, points, err