whisper: expiry refactoring

whisper/whisperv5/api.go

@@ -25,7 +25,8 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
 )
 
 var whisperOffLineErr = errors.New("whisper is offline")
@@ -64,6 +65,14 @@ func (api *PublicWhisperAPI) Version() (hexutil.Uint, error) {
 	return hexutil.Uint(api.whisper.Version()), nil
 }
 
+// Stats returns the Whisper statistics for diagnostics.
+func (api *PublicWhisperAPI) Stats() (string, error) {
+	if api.whisper == nil {
+		return "", whisperOffLineErr
+	}
+	return api.whisper.Stats(), nil
+}
+
 // MarkPeerTrusted marks specific peer trusted, which will allow it
 // to send historic (expired) messages.
 func (api *PublicWhisperAPI) MarkPeerTrusted(peerID hexutil.Bytes) error {
@@ -169,25 +178,25 @@ func (api *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (string, error) {
 
 	if len(args.Topics) == 0 && len(args.KeyName) != 0 {
 		info := "NewFilter: at least one topic must be specified"
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return "", errors.New(info)
 	}
 
 	if len(args.KeyName) != 0 && len(filter.KeySym) == 0 {
 		info := "NewFilter: key was not found by name: " + args.KeyName
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return "", errors.New(info)
 	}
 
 	if len(args.To) == 0 && len(filter.KeySym) == 0 {
 		info := "NewFilter: filter must contain either symmetric or asymmetric key"
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return "", errors.New(info)
 	}
 
 	if len(args.To) != 0 && len(filter.KeySym) != 0 {
 		info := "NewFilter: filter must not contain both symmetric and asymmetric key"
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return "", errors.New(info)
 	}
 
@@ -195,13 +204,13 @@ func (api *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (string, error) {
 		dst := crypto.ToECDSAPub(common.FromHex(args.To))
 		if !ValidatePublicKey(dst) {
 			info := "NewFilter: Invalid 'To' address"
-			log.Error(fmt.Sprintf(info))
+			glog.V(logger.Error).Infof(info)
 			return "", errors.New(info)
 		}
 		filter.KeyAsym = api.whisper.GetIdentity(string(args.To))
 		if filter.KeyAsym == nil {
 			info := "NewFilter: non-existent identity provided"
-			log.Error(fmt.Sprintf(info))
+			glog.V(logger.Error).Infof(info)
 			return "", errors.New(info)
 		}
 	}
@@ -209,7 +218,7 @@ func (api *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (string, error) {
 	if len(args.From) > 0 {
 		if !ValidatePublicKey(filter.Src) {
 			info := "NewFilter: Invalid 'From' address"
-			log.Error(fmt.Sprintf(info))
+			glog.V(logger.Error).Infof(info)
 			return "", errors.New(info)
 		}
 	}
@@ -268,13 +277,13 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error {
 		pub := crypto.ToECDSAPub(common.FromHex(args.From))
 		if !ValidatePublicKey(pub) {
 			info := "Post: Invalid 'From' address"
-			log.Error(fmt.Sprintf(info))
+			glog.V(logger.Error).Infof(info)
 			return errors.New(info)
 		}
 		params.Src = api.whisper.GetIdentity(string(args.From))
 		if params.Src == nil {
 			info := "Post: non-existent identity provided"
-			log.Error(fmt.Sprintf(info))
+			glog.V(logger.Error).Infof(info)
 			return errors.New(info)
 		}
 	}
@@ -282,7 +291,7 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error {
 	filter := api.whisper.GetFilter(args.FilterID)
 	if filter == nil && len(args.FilterID) > 0 {
 		info := fmt.Sprintf("Post: wrong filter id %s", args.FilterID)
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return errors.New(info)
 	}
 
@@ -298,7 +307,7 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error {
 			sz := len(filter.Topics)
 			if sz < 1 {
 				info := fmt.Sprintf("Post: no topics in filter # %s", args.FilterID)
-				log.Error(fmt.Sprintf(info))
+				glog.V(logger.Error).Infof(info)
 				return errors.New(info)
 			} else if sz == 1 {
 				params.Topic = filter.Topics[0]
@@ -313,26 +322,26 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error {
 	// validate
 	if len(args.KeyName) != 0 && len(params.KeySym) == 0 {
 		info := "Post: key was not found by name: " + args.KeyName
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return errors.New(info)
 	}
 
 	if len(args.To) == 0 && len(params.KeySym) == 0 {
 		info := "Post: message must be encrypted either symmetrically or asymmetrically"
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return errors.New(info)
 	}
 
 	if len(args.To) != 0 && len(params.KeySym) != 0 {
 		info := "Post: ambigous encryption method requested"
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return errors.New(info)
 	}
 
 	if len(args.To) > 0 {
 		if !ValidatePublicKey(params.Dst) {
 			info := "Post: Invalid 'To' address"
-			log.Error(fmt.Sprintf(info))
+			glog.V(logger.Error).Infof(info)
 			return errors.New(info)
 		}
 	}
@@ -341,17 +350,17 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error {
 	message := NewSentMessage(&params)
 	envelope, err := message.Wrap(&params)
 	if err != nil {
-		log.Error(fmt.Sprintf(err.Error()))
+		glog.V(logger.Error).Infof(err.Error())
 		return err
 	}
 	if len(envelope.Data) > MaxMessageLength {
 		info := "Post: message is too big"
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return errors.New(info)
 	}
 	if (envelope.Topic == TopicType{} && envelope.IsSymmetric()) {
 		info := "Post: topic is missing for symmetric encryption"
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return errors.New(info)
 	}
 

whisper/whisperv5/filter.go

@@ -18,12 +18,13 @@ package whisperv5
 
 import (
 	"crypto/ecdsa"
-	"crypto/rand"
+	crand "crypto/rand"
 	"fmt"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
 )
 
 type Filter struct {
@@ -55,7 +56,7 @@ func NewFilters(w *Whisper) *Filters {
 func (fs *Filters) generateRandomID() (id string, err error) {
 	buf := make([]byte, 20)
 	for i := 0; i < 3; i++ {
-		_, err = rand.Read(buf)
+		_, err = crand.Read(buf)
 		if err != nil {
 			continue
 		}
@@ -106,7 +107,7 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	var msg *ReceivedMessage
 	for j, watcher := range fs.watchers {
 		if p2pMessage && !watcher.AcceptP2P {
-			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), j))
+			glog.V(logger.Detail).Infof("msg [%x], filter [%d]: p2p messages are not allowed \n", env.Hash(), j)
 			continue
 		}
 
@@ -118,10 +119,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 			if match {
 				msg = env.Open(watcher)
 				if msg == nil {
-					log.Trace(fmt.Sprintf("msg [%x], filter [%s]: failed to open", env.Hash(), j))
+					glog.V(logger.Detail).Infof("msg [%x], filter [%d]: failed to open \n", env.Hash(), j)
 				}
 			} else {
-				log.Trace(fmt.Sprintf("msg [%x], filter [%s]: does not match", env.Hash(), j))
+				glog.V(logger.Detail).Infof("msg [%x], filter [%d]: does not match \n", env.Hash(), j)
 			}
 		}
 

whisper/whisperv5/message.go

@@ -30,8 +30,8 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/crypto/ecies"
+	"github.com/ethereum/go-ethereum/logger"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/logger/glog"
 	"golang.org/x/crypto/pbkdf2"
 )
 
@@ -143,7 +143,7 @@ func (msg *SentMessage) appendPadding(params *MessageParams) {
 func (msg *SentMessage) sign(key *ecdsa.PrivateKey) error {
 	if isMessageSigned(msg.Raw[0]) {
 		// this should not happen, but no reason to panic
-		log.Error(fmt.Sprintf("Trying to sign a message which was already signed"))
+		glog.V(logger.Error).Infof("Trying to sign a message which was already signed")
 		return nil
 	}
 
@@ -163,7 +163,7 @@ func (msg *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error {
 	if !ValidatePublicKey(key) {
 		return fmt.Errorf("Invalid public key provided for asymmetric encryption")
 	}
-	encrypted, err := ecies.Encrypt(crand.Reader, ecies.ImportECDSAPublic(key), msg.Raw, nil, nil)
+	encrypted, err := crypto.Encrypt(key, msg.Raw)
 	if err == nil {
 		msg.Raw = encrypted
 	}
@@ -237,7 +237,7 @@ func (msg *SentMessage) Wrap(options *MessageParams) (envelope *Envelope, err er
 		}
 	}
 	if len(msg.Raw) > MaxMessageLength {
-		log.Error(fmt.Sprintf("Message size must not exceed %d bytes", MaxMessageLength))
+		glog.V(logger.Error).Infof("Message size must not exceed %d bytes", MaxMessageLength)
 		return nil, errors.New("Oversized message")
 	}
 	var salt, nonce []byte
@@ -280,7 +280,7 @@ func (msg *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []by
 	}
 	if len(nonce) != aesgcm.NonceSize() {
 		info := fmt.Sprintf("Wrong AES nonce size - want: %d, got: %d", len(nonce), aesgcm.NonceSize())
-		log.Error(fmt.Sprintf(info))
+		glog.V(logger.Error).Infof(info)
 		return errors.New(info)
 	}
 	decrypted, err := aesgcm.Open(nil, nonce, msg.Raw, nil)
@@ -293,7 +293,7 @@ func (msg *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []by
 
 // decryptAsymmetric decrypts an encrypted payload with a private key.
 func (msg *ReceivedMessage) decryptAsymmetric(key *ecdsa.PrivateKey) error {
-	decrypted, err := ecies.ImportECDSA(key).Decrypt(crand.Reader, msg.Raw, nil, nil)
+	decrypted, err := crypto.Decrypt(key, msg.Raw)
 	if err == nil {
 		msg.Raw = decrypted
 	}
@@ -351,7 +351,7 @@ func (msg *ReceivedMessage) SigToPubKey() *ecdsa.PublicKey {
 
 	pub, err := crypto.SigToPub(msg.hash(), msg.Signature)
 	if err != nil {
-		log.Error(fmt.Sprintf("Could not get public key from signature: %v", err))
+		glog.V(logger.Error).Infof("Could not get public key from signature: %v", err)
 		return nil
 	}
 	return pub

whisper/whisperv5/peer.go

@@ -21,7 +21,8 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rlp"
 	set "gopkg.in/fatih/set.v0"
@@ -55,13 +56,13 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
 // into the network.
 func (p *Peer) start() {
 	go p.update()
-	log.Debug(fmt.Sprintf("%v: whisper started", p.peer))
+	glog.V(logger.Debug).Infof("%v: whisper started", p.peer)
 }
 
 // stop terminates the peer updater, stopping message forwarding to it.
 func (p *Peer) stop() {
 	close(p.quit)
-	log.Debug(fmt.Sprintf("%v: whisper stopped", p.peer))
+	glog.V(logger.Debug).Infof("%v: whisper stopped", p.peer)
 }
 
 // handshake sends the protocol initiation status message to the remote peer and
@@ -110,7 +111,7 @@ func (p *Peer) update() {
 
 		case <-transmit.C:
 			if err := p.broadcast(); err != nil {
-				log.Info(fmt.Sprintf("%v: broadcast failed: %v", p.peer, err))
+				glog.V(logger.Info).Infof("%v: broadcast failed: %v", p.peer, err)
 				return
 			}
 
@@ -133,20 +134,14 @@ func (peer *Peer) marked(envelope *Envelope) bool {
 // expire iterates over all the known envelopes in the host and removes all
 // expired (unknown) ones from the known list.
 func (peer *Peer) expire() {
-	// Assemble the list of available envelopes
-	available := set.NewNonTS()
-	for _, envelope := range peer.host.Envelopes() {
-		available.Add(envelope.Hash())
-	}
-	// Cross reference availability with known status
 	unmark := make(map[common.Hash]struct{})
 	peer.known.Each(func(v interface{}) bool {
-		if !available.Has(v.(common.Hash)) {
+		if !peer.host.isEnvelopeCached(v.(common.Hash)) {
 			unmark[v.(common.Hash)] = struct{}{}
 		}
 		return true
 	})
-	// Dump all known but unavailable
+	// Dump all known but no longer cached
 	for hash := range unmark {
 		peer.known.Remove(hash)
 	}
@@ -171,7 +166,7 @@ func (p *Peer) broadcast() error {
 	if err := p2p.Send(p.ws, messagesCode, transmit); err != nil {
 		return err
 	}
-	log.Trace(fmt.Sprint(p.peer, "broadcasted", len(transmit), "message(s)"))
+	glog.V(logger.Detail).Infoln(p.peer, "broadcasted", len(transmit), "message(s)")
 	return nil
 }
 

whisper/whisperv5/peer_test.go

@@ -107,7 +107,8 @@ func TestSimulation(t *testing.T) {
 }
 
 func initialize(t *testing.T) {
-	// log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat())))
+	//glog.SetV(6)
+	//glog.SetToStderr(true)
 
 	var err error
 	ip := net.IPv4(127, 0, 0, 1)

whisper/whisperv5/whisper.go

@@ -28,13 +28,20 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
 	"golang.org/x/crypto/pbkdf2"
 	set "gopkg.in/fatih/set.v0"
 )
 
+type Statistics struct {
+	messagesCleared int
+	memoryCleared   int
+	totalMemoryUsed int
+}
+
 // Whisper represents a dark communication interface through the Ethereum
 // network, using its very own P2P communication layer.
 type Whisper struct {
@@ -59,6 +66,8 @@ type Whisper struct {
 	p2pMsgQueue  chan *Envelope
 	quit         chan struct{}
 
+	stats Statistics
+
 	overflow bool
 	test     bool
 }
@@ -287,13 +296,14 @@ func (w *Whisper) Unwatch(id string) {
 // Send injects a message into the whisper send queue, to be distributed in the
 // network in the coming cycles.
 func (w *Whisper) Send(envelope *Envelope) error {
-	return w.add(envelope)
+	_, err := w.add(envelope)
+	return err
 }
 
 // Start implements node.Service, starting the background data propagation thread
 // of the Whisper protocol.
 func (w *Whisper) Start(*p2p.Server) error {
-	log.Info(fmt.Sprint("Whisper started"))
+	glog.V(logger.Info).Infoln("Whisper started")
 	go w.update()
 
 	numCPU := runtime.NumCPU()
@@ -308,7 +318,7 @@ func (w *Whisper) Start(*p2p.Server) error {
 // of the Whisper protocol.
 func (w *Whisper) Stop() error {
 	close(w.quit)
-	log.Info(fmt.Sprint("Whisper stopped"))
+	glog.V(logger.Info).Infoln("Whisper stopped")
 	return nil
 }
 
@@ -350,21 +360,24 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 		switch packet.Code {
 		case statusCode:
 			// this should not happen, but no need to panic; just ignore this message.
-			log.Warn(fmt.Sprintf("%v: unxepected status message received", p.peer))
+			glog.V(logger.Warn).Infof("%v: unxepected status message received", p.peer)
 		case messagesCode:
 			// decode the contained envelopes
 			var envelopes []*Envelope
 			if err := packet.Decode(&envelopes); err != nil {
-				log.Warn(fmt.Sprintf("%v: failed to decode envelope: [%v], peer will be disconnected", p.peer, err))
+				glog.V(logger.Warn).Infof("%v: failed to decode envelope: [%v], peer will be disconnected", p.peer, err)
 				return fmt.Errorf("garbage received")
 			}
 			// inject all envelopes into the internal pool
 			for _, envelope := range envelopes {
-				if err := wh.add(envelope); err != nil {
+				cached, err := wh.add(envelope)
-					log.Warn(fmt.Sprintf("%v: bad envelope received: [%v], peer will be disconnected", p.peer, err))
+				if err != nil {
+					glog.V(logger.Warn).Infof("%v: bad envelope received: [%v], peer will be disconnected", p.peer, err)
 					return fmt.Errorf("invalid envelope")
 				}
-				p.mark(envelope)
+				if cached {
+					p.mark(envelope)
+				}
 			}
 		case p2pCode:
 			// peer-to-peer message, sent directly to peer bypassing PoW checks, etc.
@@ -374,7 +387,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 			if p.trusted {
 				var envelope Envelope
 				if err := packet.Decode(&envelope); err != nil {
-					log.Warn(fmt.Sprintf("%v: failed to decode direct message: [%v], peer will be disconnected", p.peer, err))
+					glog.V(logger.Warn).Infof("%v: failed to decode direct message: [%v], peer will be disconnected", p.peer, err)
 					return fmt.Errorf("garbage received (directMessage)")
 				}
 				wh.postEvent(&envelope, true)
@@ -384,7 +397,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 			if wh.mailServer != nil {
 				var request Envelope
 				if err := packet.Decode(&request); err != nil {
-					log.Warn(fmt.Sprintf("%v: failed to decode p2p request message: [%v], peer will be disconnected", p.peer, err))
+					glog.V(logger.Warn).Infof("%v: failed to decode p2p request message: [%v], peer will be disconnected", p.peer, err)
 					return fmt.Errorf("garbage received (p2p request)")
 				}
 				wh.mailServer.DeliverMail(p, &request)
@@ -401,13 +414,13 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 // add inserts a new envelope into the message pool to be distributed within the
 // whisper network. It also inserts the envelope into the expiration pool at the
 // appropriate time-stamp. In case of error, connection should be dropped.
-func (wh *Whisper) add(envelope *Envelope) error {
+func (wh *Whisper) add(envelope *Envelope) (bool, error) {
 	now := uint32(time.Now().Unix())
 	sent := envelope.Expiry - envelope.TTL
 
 	if sent > now {
 		if sent-SynchAllowance > now {
-			return fmt.Errorf("envelope created in the future [%x]", envelope.Hash())
+			return false, fmt.Errorf("envelope created in the future [%x]", envelope.Hash())
 		} else {
 			// recalculate PoW, adjusted for the time difference, plus one second for latency
 			envelope.calculatePoW(sent - now + 1)
@@ -416,34 +429,34 @@ func (wh *Whisper) add(envelope *Envelope) error {
 
 	if envelope.Expiry < now {
 		if envelope.Expiry+SynchAllowance*2 < now {
-			return fmt.Errorf("very old message")
+			return false, fmt.Errorf("very old message")
 		} else {
-			log.Debug(fmt.Sprintf("expired envelope dropped [%x]", envelope.Hash()))
+			glog.V(logger.Debug).Infof("expired envelope dropped [%x]", envelope.Hash())
-			return nil // drop envelope without error
+			return false, nil // drop envelope without error
 		}
 	}
 
 	if len(envelope.Data) > MaxMessageLength {
-		return fmt.Errorf("huge messages are not allowed [%x]", envelope.Hash())
+		return false, fmt.Errorf("huge messages are not allowed [%x]", envelope.Hash())
 	}
 
 	if len(envelope.Version) > 4 {
-		return fmt.Errorf("oversized version [%x]", envelope.Hash())
+		return false, fmt.Errorf("oversized version [%x]", envelope.Hash())
 	}
 
 	if len(envelope.AESNonce) > AESNonceMaxLength {
 		// the standard AES GSM nonce size is 12,
 		// but const gcmStandardNonceSize cannot be accessed directly
-		return fmt.Errorf("oversized AESNonce [%x]", envelope.Hash())
+		return false, fmt.Errorf("oversized AESNonce [%x]", envelope.Hash())
 	}
 
 	if len(envelope.Salt) > saltLength {
-		return fmt.Errorf("oversized salt [%x]", envelope.Hash())
+		return false, fmt.Errorf("oversized salt [%x]", envelope.Hash())
 	}
 
 	if envelope.PoW() < MinimumPoW && !wh.test {
-		log.Debug(fmt.Sprintf("envelope with low PoW dropped: %f [%x]", envelope.PoW(), envelope.Hash()))
+		glog.V(logger.Debug).Infof("envelope with low PoW dropped: %f [%x]", envelope.PoW(), envelope.Hash())
-		return nil // drop envelope without error
+		return false, nil // drop envelope without error
 	}
 
 	hash := envelope.Hash()
@@ -462,15 +475,16 @@ func (wh *Whisper) add(envelope *Envelope) error {
 	wh.poolMu.Unlock()
 
 	if alreadyCached {
-		log.Trace(fmt.Sprintf("whisper envelope already cached [%x]\n", envelope.Hash()))
+		glog.V(logger.Detail).Infof("whisper envelope already cached [%x]\n", envelope.Hash())
 	} else {
-		log.Trace(fmt.Sprintf("cached whisper envelope [%x]: %v\n", envelope.Hash(), envelope))
+		glog.V(logger.Detail).Infof("cached whisper envelope [%x]: %v\n", envelope.Hash(), envelope)
+		wh.stats.totalMemoryUsed += envelope.size()
 		wh.postEvent(envelope, false) // notify the local node about the new message
 		if wh.mailServer != nil {
 			wh.mailServer.Archive(envelope)
 		}
 	}
-	return nil
+	return true, nil
 }
 
 // postEvent queues the message for further processing.
@@ -495,7 +509,7 @@ func (w *Whisper) checkOverflow() {
 	if queueSize == messageQueueLimit {
 		if !w.overflow {
 			w.overflow = true
-			log.Warn(fmt.Sprint("message queue overflow"))
+			glog.V(logger.Warn).Infoln("message queue overflow")
 		}
 	} else if queueSize <= messageQueueLimit/2 {
 		if w.overflow {
@@ -545,22 +559,32 @@ func (w *Whisper) expire() {
 	w.poolMu.Lock()
 	defer w.poolMu.Unlock()
 
+	w.stats.clear()
 	now := uint32(time.Now().Unix())
-	for then, hashSet := range w.expirations {
+	for expiry, hashSet := range w.expirations {
-		// Short circuit if a future time
+		if expiry < now {
-		if then > now {
+			w.stats.messagesCleared++
-			continue
+
+			// Dump all expired messages and remove timestamp
+			hashSet.Each(func(v interface{}) bool {
+				sz := w.envelopes[v.(common.Hash)].size()
+				w.stats.memoryCleared += sz
+				w.stats.totalMemoryUsed -= sz
+				delete(w.envelopes, v.(common.Hash))
+				delete(w.messages, v.(common.Hash))
+				return true
+			})
+			w.expirations[expiry].Clear()
+			delete(w.expirations, expiry)
 		}
-		// Dump all expired messages and remove timestamp
-		hashSet.Each(func(v interface{}) bool {
-			delete(w.envelopes, v.(common.Hash))
-			delete(w.messages, v.(common.Hash))
-			return true
-		})
-		w.expirations[then].Clear()
 	}
 }
 
+func (w *Whisper) Stats() string {
+	return fmt.Sprintf("Latest expiry cycle cleared %d messages (%d bytes). Memory usage: %d bytes.",
+		w.stats.messagesCleared, w.stats.memoryCleared, w.stats.totalMemoryUsed)
+}
+
 // envelopes retrieves all the messages currently pooled by the node.
 func (w *Whisper) Envelopes() []*Envelope {
 	w.poolMu.RLock()
@@ -589,6 +613,14 @@ func (w *Whisper) Messages(id string) []*ReceivedMessage {
 	return result
 }
 
+func (w *Whisper) isEnvelopeCached(hash common.Hash) bool {
+	w.poolMu.Lock()
+	defer w.poolMu.Unlock()
+
+	_, exist := w.envelopes[hash]
+	return exist
+}
+
 func (w *Whisper) addDecryptedMessage(msg *ReceivedMessage) {
 	w.poolMu.Lock()
 	defer w.poolMu.Unlock()
@@ -596,6 +628,11 @@ func (w *Whisper) addDecryptedMessage(msg *ReceivedMessage) {
 	w.messages[msg.EnvelopeHash] = msg
 }
 
+func (s *Statistics) clear() {
+	s.memoryCleared = 0
+	s.messagesCleared = 0
+}
+
 func ValidatePublicKey(k *ecdsa.PublicKey) bool {
 	return k != nil && k.X != nil && k.Y != nil && k.X.Sign() != 0 && k.Y.Sign() != 0
 }