Repository: liftbridge-io/liftbridge
Pull request #286 (2020-11-04 22:38:03)

Implement subscription stop position
Continuing work from: https://github.com/liftbridge-io/liftbridge-api/pull/44

Opening as a draft as could do with some guidance.

To implement `StopPosition_STOP_TIMESTAMP` I'd need to find the latest offset _before_ the specified timestamp. The CommitLog interface only exposes [`OffsetForTimestamp()`](https://github.com/liftbridge-io/liftbridge/blob/master/server/commitlog/interface.go#L27) which returns the earliest offset _after_ a timestamp.

Looks like I'd have to alter this interface to expose either:

```go
NearestOffsetBeforeTimestamp(timestamp int64) (int64, error)
NearestOffsetAfterTimestamp(timestamp int64) (int64, error)
```

or

```go
NearestOffsetToTimestamp(timestamp int64, before bool) (int64, error)
```


Patch:
diff --git a/go.mod b/go.mod
index 9576eaf..4e87219 100644
--- a/go.mod
+++ b/go.mod
@@ -8,15 +8,15 @@ require (
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
 	github.com/dustin/go-humanize v1.0.0
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
-	github.com/golang/protobuf v1.4.2
+	github.com/golang/protobuf v1.4.3
 	github.com/hako/durafmt v0.0.0-20200605151348-3a43fc422dd9
 	github.com/hashicorp/go-hclog v0.14.1 // indirect
 	github.com/hashicorp/go-immutable-radix v1.2.0 // indirect
 	github.com/hashicorp/go-msgpack v1.1.5 // indirect
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/hashicorp/raft v1.1.2
-	github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201026195646-f7afc15d3aec
-	github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201022201130-7ae2595b40b7
+	github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201118225953-b849cccb6467
+	github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201029165056-10f2aa65f256
 	github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3
 	github.com/liftbridge-io/raft-boltdb v0.0.0-20200414234651-aaf6e08d8f73
 	github.com/mattn/go-colorable v0.1.7 // indirect
@@ -40,7 +40,11 @@ require (
 	github.com/urfave/cli v1.22.4
 	go.etcd.io/bbolt v1.3.5 // indirect
 	golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de // indirect
-	google.golang.org/grpc v1.32.0
+	golang.org/x/net v0.0.0-20201110031124-69a78807bb2b // indirect
+	golang.org/x/sys v0.0.0-20201118182958-a01c418693c7 // indirect
+	golang.org/x/text v0.3.4 // indirect
+	google.golang.org/genproto v0.0.0-20201117123952-62d171c70ae1 // indirect
+	google.golang.org/grpc v1.33.2
 	gopkg.in/ini.v1 v1.57.0 // indirect
 	launchpad.net/gocheck v0.0.0-20140225173054-000000000087 // indirect
 )
diff --git a/go.sum b/go.sum
index 9126d92..3adff4e 100644
--- a/go.sum
+++ b/go.sum
@@ -97,6 +97,8 @@ github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvq
 github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
 github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
+github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -110,6 +112,7 @@ github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXi
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
@@ -183,10 +186,13 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201026195646-f7afc15d3aec h1:oP+ZkkSmdlLow/colBEiw3n9VG+TRF8su392KtposEw=
-github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201026195646-f7afc15d3aec/go.mod h1:4s+8qz5Y9qen5Sd6VV7JNdUTXfrPnK0O7VoqRKrTJsA=
-github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201022201130-7ae2595b40b7 h1:xrF+cDIKVXEzs3dizCDU8SCgC9prT3X562D72iU4W/I=
-github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201022201130-7ae2595b40b7/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.1 h1:a38Uyf+6HyENKPr/XifoJv9FHTmaqzwzEBoQTklE1R4=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.1/go.mod h1:ywA1518j+cuorMseepnLfECQNBstC0HGsibELAnMWpw=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201118225953-b849cccb6467 h1:Ls62H5vuS7lNfBdDnLaA0kG8hIa91fNJRQY2WGYV/aM=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201118225953-b849cccb6467/go.mod h1:24NMu02Ba2sMO2y+IYstP1UFKVA4a6/p54Lc7KccSLc=
+github.com/liftbridge-io/liftbridge-api v1.0.1-0.20200703154436-52124ef6c36f/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
+github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201029165056-10f2aa65f256 h1:2pZtC3v6IBTwE70xfb/k0DPlOJ6BlXpthCUWrxCnhwo=
+github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201029165056-10f2aa65f256/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
 github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3 h1:O4mg1NEmukgY8hxen3grrG5RY34LadMTzpbjf8kM2tA=
 github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3/go.mod h1:wmIIYVq+psahPlB1rvtTkGiltdihsKJbqwE1DkIPwj4=
 github.com/liftbridge-io/raft-boltdb v0.0.0-20200414234651-aaf6e08d8f73 h1:8r/ReB1ns87pVDwSnPj87HIbOu/5y0uDyGChx9mUGSQ=
@@ -375,8 +381,12 @@ golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200625001655-4c5254603344 h1:vGXIOMxbNfDTk/aXCmfdLgkrSV+Z2tcbze+pEc3v5W4=
+golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201002202402-0a1ea396d57c h1:dk0ukUIHmGHqASjP0iue2261isepFCC6XRCSd1nHgDw=
 golang.org/x/net v0.0.0-20201002202402-0a1ea396d57c/go.mod h1:iQL9McJNjoIa5mjH6nYTCTZXUN6RP+XW3eib7Ya3XcI=
+golang.org/x/net v0.0.0-20201110031124-69a78807bb2b h1:uwuIcX0g4Yl1NC5XAz37xsr2lTtcqevgzYNVt49waME=
+golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -412,13 +422,20 @@ golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae h1:Ih9Yo4hSPImZOpfGuA4bR/ORKTAbhZo2AbWNRCnevdo=
+golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201118182958-a01c418693c7 h1:Z991aAXPjz0tLnj74pVXW3eWJ5lHMIBvbRfMq4M2jHA=
+golang.org/x/sys v0.0.0-20201118182958-a01c418693c7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.4 h1:0YWbFKbhXG/wIiuHDSKpS0Iy7FSA+u45VtBMfQcFTTc=
+golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -462,16 +479,23 @@ google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98
 google.golang.org/genproto v0.0.0-20190911173649-1774047e7e51/go.mod h1:IbNlFCBrqXvoKpeg0TB2l7cyZUmoaFKYIwrEpbDKLA8=
 google.golang.org/genproto v0.0.0-20191108220845-16a3f7862a1a/go.mod h1:n3cpQtvxv34hfy77yVDNjmbRyujviMdxYliBSkLhpCc=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
+google.golang.org/genproto v0.0.0-20200702021140-07506425bd67 h1:4BC1C1i30F3MZeiIO6y6IIo4DxrtOwITK87bQl3lhFA=
+google.golang.org/genproto v0.0.0-20200702021140-07506425bd67/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20201002142447-3860012362da h1:DTQYk4u7nICKkkVZsBv0/0po0ChISxAJ5CTAfUhO0PQ=
 google.golang.org/genproto v0.0.0-20201002142447-3860012362da/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
+google.golang.org/genproto v0.0.0-20201117123952-62d171c70ae1 h1:EVow1AaDgdoMjdO64/fntn4+RGTVor8YE/mkmIYsqFM=
+google.golang.org/genproto v0.0.0-20201117123952-62d171c70ae1/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
+google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.32.0 h1:zWTV+LMdc3kaiJMSTOFz2UgSBgx8RNQoTGiZu3fR9S0=
 google.golang.org/grpc v1.32.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
+google.golang.org/grpc v1.33.2 h1:EQyQC3sa8M+p6Ulc8yy9SWSS2GVwyRc83gAbG8lrl4o=
+google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
diff --git a/server/api.go b/server/api.go
index 64c8128..975519b 100644
--- a/server/api.go
+++ b/server/api.go
@@ -590,9 +590,14 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 		return nil, nil, st
 	}
 
-	stopOffset := waitForNewMessages
-	if partition.IsReadonly() {
-		stopOffset = partition.log.NewestOffset()
+	stopOffset, st := getStopOffset(req, partition.log)
+	if st != nil {
+		return nil, nil, st
+	}
+
+	if stopOffset != waitForNewMessages && stopOffset < startOffset {
+		return nil, nil, status.New(
+			codes.InvalidArgument, fmt.Sprintf("Stop offset is before start offset: %d < %d", stopOffset, startOffset))
 	}
 
 	var (
@@ -628,7 +633,7 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 					s = status.New(code, err.Error())
 				} else if err == commitlog.ErrCommitLogReadonly {
 					// Partition was set to readonly while subscribed.
-					s = status.New(codes.ResourceExhausted, fmt.Sprintf("End of readonly partition"))
+					s = status.New(codes.ResourceExhausted, "End of readonly partition")
 				} else {
 					s = status.Convert(err)
 				}
@@ -658,8 +663,8 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 			case <-cancel:
 				return
 			}
-			if stopOffset != waitForNewMessages && offset == stopOffset {
-				s := status.New(codes.ResourceExhausted, fmt.Sprintf("Stop offset reached"))
+			if offset == stopOffset {
+				s := status.New(codes.ResourceExhausted, "Stop offset reached")
 
 				select {
 				case errCh <- s:
@@ -679,7 +684,7 @@ func getStartOffset(req *client.SubscribeRequest, log commitlog.CommitLog) (int6
 	case client.StartPosition_OFFSET:
 		startOffset = req.StartOffset
 	case client.StartPosition_TIMESTAMP:
-		offset, err := log.OffsetForTimestamp(req.StartTimestamp)
+		offset, err := log.EarliestOffsetAfterTimestamp(req.StartTimestamp)
 		if err != nil {
 			return startOffset, status.New(
 				codes.Internal, fmt.Sprintf("Failed to lookup offset for timestamp: %v", err))
@@ -705,6 +710,37 @@ func getStartOffset(req *client.SubscribeRequest, log commitlog.CommitLog) (int6
 	return startOffset, nil
 }
 
+func getStopOffset(req *client.SubscribeRequest, log commitlog.CommitLog) (int64, *status.Status) {
+	var stopOffset int64
+	switch req.StopPosition {
+	case client.StopPosition_STOP_ON_CANCEL:
+		stopOffset = waitForNewMessages
+		if log.IsReadonly() {
+			stopOffset = log.NewestOffset()
+		}
+	case client.StopPosition_STOP_OFFSET:
+		stopOffset = req.StopOffset
+	case client.StopPosition_STOP_TIMESTAMP:
+		var err error
+		stopOffset, err = log.LatestOffsetBeforeTimestamp(req.StopTimestamp)
+		if err != nil {
+			return stopOffset, status.New(
+				codes.Internal, fmt.Sprintf("Failed to lookup offset for timestamp: %v", err))
+		}
+	case client.StopPosition_STOP_LATEST:
+		stopOffset = log.NewestOffset()
+		if stopOffset == -1 {
+			return stopOffset, status.New(codes.ResourceExhausted, "Stream is empty")
+		}
+	default:
+		return stopOffset, status.New(
+			codes.InvalidArgument,
+			fmt.Sprintf("Unknown StopPosition %s", req.StopPosition))
+	}
+
+	return stopOffset, nil
+}
+
 func getStreamConfig(req *client.CreateStreamRequest) *proto.StreamConfig {
 	config := new(proto.StreamConfig)
 	if req.RetentionMaxAge != nil {
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index 0bd76d8..f6f1134 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -285,9 +285,9 @@ func (l *commitLog) OldestOffset() int64 {
 	return l.segments[0].FirstOffset()
 }
 
-// OffsetForTimestamp returns the earliest offset whose timestamp is greater
-// than or equal to the given timestamp.
-func (l *commitLog) OffsetForTimestamp(timestamp int64) (int64, error) {
+// EarliestOffsetAfterTimestamp returns the earliest offset whose timestamp is
+// greater than or equal to the given timestamp.
+func (l *commitLog) EarliestOffsetAfterTimestamp(timestamp int64) (int64, error) {
 	l.mu.RLock()
 	defer l.mu.RUnlock()
 
@@ -328,6 +328,52 @@ func (l *commitLog) OffsetForTimestamp(timestamp int64) (int64, error) {
 	return l.segments[len(l.segments)-1].NextOffset(), nil
 }
 
+// LatestOffsetBeforeTimestamp returns the latest offset whose timestamp is less
+// than or equal to the given timestamp.
+func (l *commitLog) LatestOffsetBeforeTimestamp(timestamp int64) (int64, error) {
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+
+	// Find the first segment whose base timestamp is greater than the given
+	// timestamp.
+	idx, err := findSegmentIndexByTimestamp(l.segments, timestamp)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to find log segment for timestamp")
+	}
+	// Search the previous segment for the first entry whose timestamp is
+	// greater than or equal to the given timestamp. If this is the first
+	// segment, just search it.
+	var seg *segment
+	if idx == 0 {
+		seg = l.segments[0]
+		// if the given timestamp is before the start of the stream return an
+		// error.
+		if timestamp < seg.FirstWriteTime() {
+			return 0, errors.New("timestamp is before the beginning of the log")
+		}
+	} else {
+		seg = l.segments[idx-1]
+	}
+
+	// Find entry equal to or greater than the given timestamp.
+	entry, err := seg.findEntryByTimestamp(timestamp)
+	if err == nil {
+		// If it's an exact match, return the offset.
+		if entry.Timestamp == timestamp {
+			return entry.Offset, nil
+		}
+
+		// Otherwise we want the previous offset.
+		return entry.Offset - 1, nil
+	}
+
+	if err != ErrEntryNotFound && err != io.EOF {
+		return 0, errors.Wrap(err, "failed to find log entry for timestamp")
+	}
+
+	return seg.lastOffset, nil
+}
+
 // SetHighWatermark sets the high watermark on the log. All messages up to and
 // including the high watermark are considered committed.
 func (l *commitLog) SetHighWatermark(hw int64) {
diff --git a/server/commitlog/interface.go b/server/commitlog/interface.go
index 4bfe4b7..d75fcfa 100644
--- a/server/commitlog/interface.go
+++ b/server/commitlog/interface.go
@@ -22,9 +22,13 @@ type CommitLog interface {
 	// empty.
 	OldestOffset() int64
 
-	// OffsetForTimestamp returns the earliest offset whose timestamp is
-	// greater than or equal to the given timestamp.
-	OffsetForTimestamp(timestamp int64) (int64, error)
+	// EarliestOffsetAfterTimestamp returns the earliest offset whose timestamp
+	// is greater than or equal to the given timestamp.
+	EarliestOffsetAfterTimestamp(timestamp int64) (int64, error)
+
+	// LatestOffsetBeforeTimestamp returns the latest offset whose timestamp is
+	// less than or equal to the given timestamp.
+	LatestOffsetBeforeTimestamp(timestamp int64) (int64, error)
 
 	// SetHighWatermark sets the high watermark on the log. All messages up to
 	// and including the high watermark are considered committed.
diff --git a/server/commitlog/segment.go b/server/commitlog/segment.go
index 789a0df..fbacbff 100644
--- a/server/commitlog/segment.go
+++ b/server/commitlog/segment.go
@@ -185,6 +185,12 @@ func (s *segment) FirstOffset() int64 {
 	return s.firstOffset
 }
 
+func (s *segment) FirstWriteTime() int64 {
+	s.RLock()
+	defer s.RUnlock()
+	return s.firstWriteTime
+}
+
 func (s *segment) LastOffset() int64 {
 	s.RLock()
 	defer s.RUnlock()
@@ -383,14 +389,14 @@ func (s *segment) findEntry(offset int64) (e *entry, err error) {
 }
 
 // findEntryByTimestamp returns the first entry whose timestamp is greater than
-// or equal to the given offset.
+// or equal to the given timestamp.
 func (s *segment) findEntryByTimestamp(timestamp int64) (e *entry, err error) {
 	s.RLock()
 	defer s.RUnlock()
 	e = &entry{}
-	n := int(s.Index.Position() / entryWidth)
+	n := int(s.Index.CountEntries())
 	idx := sort.Search(n, func(i int) bool {
-		if err := s.Index.ReadEntryAtFileOffset(e, int64(i*entryWidth)); err != nil {
+		if err := s.Index.ReadEntryAtLogOffset(e, int64(i)); err != nil {
 			panic(err)
 		}
 		return e.Timestamp >= timestamp
@@ -398,7 +404,7 @@ func (s *segment) findEntryByTimestamp(timestamp int64) (e *entry, err error) {
 	if idx == n {
 		return nil, ErrEntryNotFound
 	}
-	err = s.Index.ReadEntryAtFileOffset(e, int64(idx*entryWidth))
+	err = s.Index.ReadEntryAtLogOffset(e, int64(idx))
 	return e, err
 }
 
diff --git a/server/metadata.go b/server/metadata.go
index c971027..3cacbce 100644
--- a/server/metadata.go
+++ b/server/metadata.go
@@ -1332,10 +1332,10 @@ func eventTimestampsToProto(timestamps EventTimestamps) *client.PartitionEventTi
 }
 
 // getPartitionMetadata returns a partition's metadata.
-func getPartitionMetadata(partitionId int32, partition *partition) *client.PartitionMetadata {
+func getPartitionMetadata(partitionID int32, partition *partition) *client.PartitionMetadata {
 	leader, _ := partition.GetLeader()
 	return &client.PartitionMetadata{
-		Id:                         partitionId,
+		Id:                         partitionID,
 		Leader:                     leader,
 		Replicas:                   partition.GetReplicas(),
 		Isr:                        partition.GetISR(),


Test patch:
diff --git a/server/api_test.go b/server/api_test.go
index 8a0cae9..7a479a2 100644
--- a/server/api_test.go
+++ b/server/api_test.go
@@ -842,6 +842,105 @@ func TestSubscribePartitionClosed(t *testing.T) {
 	}
 }
 
+func TestSubscribeStopPosition(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Configure server.
+	s1Config := getTestConfig("a", true, 5050)
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	stream := "foo"
+	ctx := context.Background()
+	err = client.CreateStream(ctx, "foo", stream)
+	require.NoError(t, err)
+
+	// subscription to empty stream errors
+	err = client.Subscribe(ctx, stream, nil, lift.StopAtLatestReceived())
+	require.Error(t, err)
+	require.Equal(t, lift.ErrReadonlyPartition, err)
+
+	// publish messages to stream
+	numMessages := 10
+	acks := make([]*lift.Ack, 0, numMessages)
+	for i := 0; i < numMessages; i++ {
+		ack, err := client.Publish(ctx, stream, []byte(strconv.Itoa(i)))
+		require.NoError(t, err)
+		acks = append(acks, ack)
+	}
+
+	exhausted := make(chan struct{})
+
+	// stream should return all message then close
+	receivedCount := 0
+	err = client.Subscribe(ctx, stream, func(msg *lift.Message, err error) {
+		receivedCount++
+
+		// we've recieved all messages so this must be a resource exhaused error
+		if receivedCount > numMessages {
+			require.Error(t, err)
+			require.Equal(t, lift.ErrReadonlyPartition, err)
+			exhausted <- struct{}{}
+		}
+	},
+		lift.StartAtEarliestReceived(),
+		lift.StopAtLatestReceived(),
+	)
+	require.NoError(t, err)
+
+	select {
+	case <-exhausted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Stream was not exhausted")
+	}
+
+	// Get all messages between the timestamps of two already acknowledged
+	// messages. Inclusive of the bounds:
+	// 0, 1, 2, 3, 4, 5, 6, 7, 8, 9
+	//      |_____________|
+	start := acks[2]
+	stop := acks[6]
+	expectedCount := 5
+	expectedOffset := start.Offset()
+
+	receivedCount = 0
+	err = client.Subscribe(ctx, stream, func(msg *lift.Message, err error) {
+		receivedCount++
+
+		if msg != nil {
+			require.Equal(t, expectedOffset, msg.Offset())
+			expectedOffset++
+		}
+
+		// we've recieved all messages so this must be a resource exhaused error
+		if receivedCount > expectedCount {
+			require.Error(t, err)
+			require.Equal(t, lift.ErrReadonlyPartition, err)
+			exhausted <- struct{}{}
+		}
+	},
+		lift.StartAtTime(start.ReceptionTimestamp()),
+		lift.StopAtTime(stop.ReceptionTimestamp()),
+	)
+	require.NoError(t, err)
+
+	select {
+	case <-exhausted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Stream was not exhausted")
+	}
+}
+
 // Ensure getStreamConfig applies non-nil values from the CreateStreamRequest
 // to the StreamConfig.
 func TestGetStreamConfig(t *testing.T) {
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index 390f107..713db43 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -372,9 +372,9 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 	require.Equal(t, int64(14), l.LastOffsetForLeaderEpoch(3))
 }
 
-// Ensures OffsetForTimestamp returns the earliest offset whose timestamp is
-// greater than or equal to the given timestamp.
-func TestOffsetForTimestamp(t *testing.T) {
+// Ensures EarliestOffsetAfterTimestamp returns the earliest offset whose
+// timestamp is greater than or equal to the given timestamp.
+func TestEarliestOffsetAfterTimestamp(t *testing.T) {
 	opts := Options{
 		Path:            tempDir(t),
 		MaxSegmentBytes: 100,
@@ -394,37 +394,94 @@ func TestOffsetForTimestamp(t *testing.T) {
 	}
 
 	// Underflowed timestamp should return the first offset.
-	offset, err := l.OffsetForTimestamp(-1)
+	offset, err := l.EarliestOffsetAfterTimestamp(-1)
 	require.NoError(t, err)
 	require.Equal(t, int64(0), offset)
 
 	// Find offset in the first segment.
-	offset, err = l.OffsetForTimestamp(20)
+	offset, err = l.EarliestOffsetAfterTimestamp(20)
 	require.NoError(t, err)
 	require.Equal(t, int64(2), offset)
 
 	// Find offset in an inner segment.
-	offset, err = l.OffsetForTimestamp(30)
+	offset, err = l.EarliestOffsetAfterTimestamp(30)
 	require.NoError(t, err)
 	require.Equal(t, int64(3), offset)
 
 	// Find offset in the last segment.
-	offset, err = l.OffsetForTimestamp(90)
+	offset, err = l.EarliestOffsetAfterTimestamp(90)
 	require.NoError(t, err)
 	require.Equal(t, int64(9), offset)
 
 	// Overflowed timestamp should return the next offset.
-	offset, err = l.OffsetForTimestamp(500)
+	offset, err = l.EarliestOffsetAfterTimestamp(500)
 	require.NoError(t, err)
 	require.Equal(t, int64(10), offset)
 
 	// Find offset for timestamp not present in log. Should return offset with
 	// next highest timestamp.
-	offset, err = l.OffsetForTimestamp(25)
+	offset, err = l.EarliestOffsetAfterTimestamp(25)
 	require.NoError(t, err)
 	require.Equal(t, int64(3), offset)
 }
 
+// Ensures LatestOffsetBeforeTimestamp returns the latest offset whose
+// timestamp is less than or equal to the given timestamp.
+func TestLatestOffsetBeforeTimestamp(t *testing.T) {
+	opts := Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 100,
+	}
+	l, cleanup := setupWithOptions(t, opts)
+	defer cleanup()
+
+	// Append some messages.
+	numMsgs := 10
+	msgs := make([]*Message, numMsgs)
+	for i := 0; i < numMsgs; i++ {
+		msgs[i] = &Message{Value: []byte(strconv.Itoa(i)), Timestamp: int64(i * 10)}
+	}
+	for _, msg := range msgs {
+		_, err := l.Append([]*Message{msg})
+		require.NoError(t, err)
+	}
+
+	// Underflowed timestamp should return an error.
+	_, err := l.LatestOffsetBeforeTimestamp(-1)
+	require.Error(t, err)
+
+	// Find offset for timestamp right after first entry.
+	offset, err := l.LatestOffsetBeforeTimestamp(5)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), offset)
+
+	// Find offset in the first segment.
+	offset, err = l.LatestOffsetBeforeTimestamp(20)
+	require.NoError(t, err)
+	require.Equal(t, int64(2), offset)
+
+	// Find offset in an inner segment.
+	offset, err = l.LatestOffsetBeforeTimestamp(30)
+	require.NoError(t, err)
+	require.Equal(t, int64(3), offset)
+
+	// Find offset in the last segment.
+	offset, err = l.LatestOffsetBeforeTimestamp(90)
+	require.NoError(t, err)
+	require.Equal(t, int64(9), offset)
+
+	// Overflowed timestamp should return high water mark.
+	offset, err = l.LatestOffsetBeforeTimestamp(500)
+	require.NoError(t, err)
+	require.Equal(t, int64(9), offset)
+
+	// Find offset for timestamp not present in log. Should return offset with
+	// next lowest timestamp.
+	offset, err = l.LatestOffsetBeforeTimestamp(25)
+	require.NoError(t, err)
+	require.Equal(t, int64(2), offset)
+}
+
 // Ensure Truncate removes log entries up to the given offset and that the
 // leader epoch cache is also truncated.
 func TestTruncate(t *testing.T) {