diff --git a/akka-persistence/akka-persistence-common/src/main/scala/Storage.scala b/akka-persistence/akka-persistence-common/src/main/scala/Storage.scala
index 379dd0b5f4..c11541accc 100644
--- a/akka-persistence/akka-persistence-common/src/main/scala/Storage.scala
+++ b/akka-persistence/akka-persistence-common/src/main/scala/Storage.scala
@@ -218,9 +218,9 @@ trait PersistentVector[T] extends IndexedSeq[T] with Transactional with Committa
     else storage.getVectorStorageEntryFor(uuid, index)
   }
 
-  override def slice(start: Int, count: Int): IndexedSeq[T] = slice(Some(start), None, count)
+  override def slice(start: Int, finish: Int): IndexedSeq[T] = slice(Some(start), Some(finish))
   
-  def slice(start: Option[Int], finish: Option[Int], count: Int): IndexedSeq[T] = {
+  def slice(start: Option[Int], finish: Option[Int], count: Int = 0): IndexedSeq[T] = {
     val buffer = new scala.collection.mutable.ArrayBuffer[T]
     storage.getVectorStorageRangeFor(uuid, start, finish, count).foreach(buffer.append(_))
     buffer
diff --git a/akka-persistence/akka-persistence-mongo/src/main/scala/MongoStorageBackend.scala b/akka-persistence/akka-persistence-mongo/src/main/scala/MongoStorageBackend.scala
index cfe996be06..97245cff02 100644
--- a/akka-persistence/akka-persistence-mongo/src/main/scala/MongoStorageBackend.scala
+++ b/akka-persistence/akka-persistence-mongo/src/main/scala/MongoStorageBackend.scala
@@ -248,9 +248,17 @@ private[akka] object MongoStorageBackend extends
           dbo.get(VALUE).asInstanceOf[JList[AnyRef]]
       }
 
+      val s = if (start.isDefined) start.get else 0
+      val cnt =
+        if (finish.isDefined) {
+          val f = finish.get
+          if (f >= s) (f - s) else count
+        }
+        else count
+
       // pick the subrange and make a Scala list
       val l =
-        List(o.subList(start.get, start.get + count).toArray: _*)
+        List(o.subList(s, s + cnt).toArray: _*)
 
       for(e <- l)
         yield serializer.in[AnyRef](e.asInstanceOf[Array[Byte]])
diff --git a/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala b/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala
index 93aa1862d1..eb231ad473 100644
--- a/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala
+++ b/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala
@@ -1,9 +1,8 @@
 package se.scalablesolutions.akka.persistence.mongo
 
-import junit.framework.TestCase
-
 import org.junit.{Test, Before}
 import org.junit.Assert._
+import org.scalatest.junit.JUnitSuite
 
 import _root_.dispatch.json.{JsNumber, JsValue}
 import _root_.dispatch.json.Js._
@@ -27,6 +26,7 @@ case class Balance(accountNo: String)
 case class Debit(accountNo: String, amount: BigInt, failer: Actor)
 case class MultiDebit(accountNo: String, amounts: List[BigInt], failer: Actor)
 case class Credit(accountNo: String, amount: BigInt)
+case class Log(start: Int, finish: Int)
 case object LogSize
 
 class BankAccountActor extends Transactor {
@@ -88,6 +88,9 @@ class BankAccountActor extends Transactor {
 
     case LogSize =>
       reply(txnLog.length.asInstanceOf[AnyRef])
+
+    case Log(start, finish) =>
+      reply(txnLog.slice(start, finish))
   }
 }
 
@@ -98,7 +101,7 @@ class BankAccountActor extends Transactor {
   }
 }
 
-class MongoPersistentActorSpec extends TestCase {
+class MongoPersistentActorSpec extends JUnitSuite {
   @Test
   def testSuccessfulDebit = {
     val bactor = new BankAccountActor
@@ -121,7 +124,14 @@ class MongoPersistentActorSpec extends TestCase {
     val JsNumber(b2) = (bactor !! Balance("a-123")).get.asInstanceOf[JsValue] 
     assertEquals(BigInt(1000), BigInt(b2.intValue))
 
-    assertEquals(7, (bactor !! LogSize).get)
+    assert(7 == (bactor !! LogSize).get.asInstanceOf[Int])
+
+    import scala.collection.mutable.ArrayBuffer
+    assert((bactor !! Log(0, 7)).get.asInstanceOf[ArrayBuffer[String]].size == 7)
+    assert((bactor !! Log(0, 0)).get.asInstanceOf[ArrayBuffer[String]].size == 0)
+    assert((bactor !! Log(1, 2)).get.asInstanceOf[ArrayBuffer[String]].size == 1)
+    assert((bactor !! Log(6, 7)).get.asInstanceOf[ArrayBuffer[String]].size == 1)
+    assert((bactor !! Log(0, 1)).get.asInstanceOf[ArrayBuffer[String]].size == 1)
   }
 
   @Test
@@ -144,7 +154,7 @@ class MongoPersistentActorSpec extends TestCase {
     assertEquals(BigInt(5000), BigInt(b1.intValue))
 
     // should not count the failed one
-    assertEquals(3, (bactor !! LogSize).get)
+    assert(3 == (bactor !! LogSize).get.asInstanceOf[Int])
   }
 
   @Test
@@ -167,6 +177,6 @@ class MongoPersistentActorSpec extends TestCase {
     assertEquals(BigInt(5000), BigInt(b1.intValue))
 
     // should not count the failed one
-    assertEquals(3, (bactor !! LogSize).get)
+    assert(3 == (bactor !! LogSize).get.asInstanceOf[Int])
   }
 }
diff --git a/akka-persistence/akka-persistence-mongo/src/test/scala/MongoStorageSpec.scala b/akka-persistence/akka-persistence-mongo/src/test/scala/MongoStorageSpec.scala
index 97307dde17..ae2a1e4dd4 100644
--- a/akka-persistence/akka-persistence-mongo/src/test/scala/MongoStorageSpec.scala
+++ b/akka-persistence/akka-persistence-mongo/src/test/scala/MongoStorageSpec.scala
@@ -1,20 +1,19 @@
 package se.scalablesolutions.akka.persistence.mongo
 
-import junit.framework.TestCase
-
 import org.junit.{Test, Before}
 import org.junit.Assert._
+import org.scalatest.junit.JUnitSuite
 import _root_.dispatch.json._
 import _root_.dispatch.json.Js._
 import java.util.NoSuchElementException
 
 @scala.reflect.BeanInfo case class Foo(no: Int, name: String)
-class MongoStorageSpec extends TestCase {
+class MongoStorageSpec extends JUnitSuite {
 
   val changeSetV = new scala.collection.mutable.ArrayBuffer[AnyRef]
   val changeSetM = new scala.collection.mutable.HashMap[AnyRef, AnyRef]
 
-  override def setUp = {
+  @Before def initialize() = {
     MongoStorageBackend.coll.drop
   }
 
diff --git a/akka-persistence/akka-persistence-redis/src/main/scala/RedisStorageBackend.scala b/akka-persistence/akka-persistence-redis/src/main/scala/RedisStorageBackend.scala
index 79aae70fd1..8e2adaa5c3 100644
--- a/akka-persistence/akka-persistence-redis/src/main/scala/RedisStorageBackend.scala
+++ b/akka-persistence/akka-persistence-redis/src/main/scala/RedisStorageBackend.scala
@@ -228,17 +228,21 @@ private [akka] object RedisStorageBackend extends
 
   def getVectorStorageRangeFor(name: String, start: Option[Int], finish: Option[Int], count: Int): List[Array[Byte]] = withErrorHandling {
     /**
-     * <tt>count</tt> is the max number of results to return. Start with
-     * <tt>start</tt> or 0 (if <tt>start</tt> is not defined) and go until
-     * you hit <tt>finish</tt> or <tt>count</tt>.
+     * if <tt>start</tt> and <tt>finish</tt> both are defined, ignore <tt>count</tt> and
+     * report the range [start, finish)
+     * if <tt>start</tt> is not defined, assume <tt>start</tt> = 0
+     * if <tt>start</tt> == 0 and <tt>finish</tt> == 0, return an empty collection
      */
     val s = if (start.isDefined) start.get else 0
     val cnt =
       if (finish.isDefined) {
         val f = finish.get
-        if (f >= s) Math.min(count, (f - s)) else count
+        // if (f >= s) Math.min(count, (f - s)) else count
+        if (f >= s) (f - s) else count
       }
       else count
+    if (s == 0 && cnt == 0) List()
+    else
     db.lrange(new String(encode(name.getBytes)), s, s + cnt - 1) match {
       case None =>
         throw new NoSuchElementException(name + " does not have elements in the range specified")
diff --git a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala
index f21b1b72a0..d27ec1bd31 100644
--- a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala
+++ b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala
@@ -1,7 +1,5 @@
 package se.scalablesolutions.akka.persistence.redis
 
-import junit.framework.TestCase
-
 import org.junit.{Test, Before}
 import org.junit.Assert._
 
@@ -24,6 +22,7 @@ case class Balance(accountNo: String)
 case class Debit(accountNo: String, amount: BigInt, failer: Actor)
 case class MultiDebit(accountNo: String, amounts: List[BigInt], failer: Actor)
 case class Credit(accountNo: String, amount: BigInt)
+case class Log(start: Int, finish: Int)
 case object LogSize
 
 class AccountActor extends Transactor {
@@ -83,6 +82,9 @@ class AccountActor extends Transactor {
 
     case LogSize =>
       reply(txnLog.length.asInstanceOf[AnyRef])
+
+    case Log(start, finish) =>
+      reply(txnLog.slice(start, finish))
   }
 }
 
@@ -94,7 +96,8 @@ class AccountActor extends Transactor {
   }
 }
 
-class RedisPersistentActorSpec extends TestCase {
+import org.scalatest.junit.JUnitSuite
+class RedisPersistentActorSpec extends JUnitSuite {
   @Test
   def testSuccessfulDebit = {
     val bactor = new AccountActor
@@ -113,6 +116,12 @@ class RedisPersistentActorSpec extends TestCase {
 
     val c: Int = (bactor !! LogSize).get
     assertTrue(7 == c)
+    import scala.collection.mutable.ArrayBuffer
+    assert((bactor !! Log(0, 7)).get.asInstanceOf[ArrayBuffer[String]].size == 7)
+    assert((bactor !! Log(0, 0)).get.asInstanceOf[ArrayBuffer[String]].size == 0)
+    assert((bactor !! Log(1, 2)).get.asInstanceOf[ArrayBuffer[String]].size == 1)
+    assert((bactor !! Log(6, 7)).get.asInstanceOf[ArrayBuffer[String]].size == 1)
+    assert((bactor !! Log(0, 1)).get.asInstanceOf[ArrayBuffer[String]].size == 1)
   }
 
   @Test
diff --git a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala
index 2e5b5a507e..e53c958f9b 100644
--- a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala
+++ b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala
@@ -1,7 +1,5 @@
 package se.scalablesolutions.akka.persistence.redis
 
-import junit.framework.TestCase
-
 import org.junit.{Test, Before}
 import org.junit.Assert._
 
@@ -50,7 +48,8 @@ class QueueActor extends Transactor {
   }
 }
 
-class RedisPersistentQSpec extends TestCase {
+import org.scalatest.junit.JUnitSuite
+class RedisPersistentQSpec extends JUnitSuite {
   @Test
   def testSuccessfulNQ = {
     val qa = new QueueActor