Repository: liftbridge-io/liftbridge
Pull request #153 (2020-02-14 23:38:43)

Update api
Implement the new API and protocol (https://github.com/liftbridge-io/liftbridge-api) in preparation for v1.0.0.


Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index f9114aa..bf1a695 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -32,8 +32,9 @@ jobs:
       - run:
           name: Test
           command: |
-            go test -v .
-            go test -v -coverprofile=coverage.txt ./server/...
+            go test -v . # Run tests in root but exclude from coverage
+            go test -v -coverprofile=coverage.txt.tmp ./server/...
+            cat coverage.txt.tmp | grep -v ".pb.go" > coverage.txt # Exclude protobuf from coverage
             goveralls -coverprofile=coverage.txt -service=circle-ci -repotoken=$COVERALLS_REPO_TOKEN
 
   build-and-push-standalone-dev-image:
diff --git a/documentation/client_implementation.md b/documentation/client_implementation.md
index d1b8cd1..9cb0262 100644
--- a/documentation/client_implementation.md
+++ b/documentation/client_implementation.md
@@ -11,13 +11,13 @@ inaccurate, feel free to make a PR fixing it.
 
 Liftbridge relies on gRPC and Protocol Buffers (protobufs) for its API. These
 definitions are stored in the
-[liftbridge-grpc](https://github.com/liftbridge-io/liftbridge-api) repository.
+[liftbridge-api](https://github.com/liftbridge-io/liftbridge-api) repository.
 Refer to the [gRPC documentation](https://grpc.io/docs/) for guidance on
 generating the gRPC code from the protobuf definitions for your particular
-programming language. The liftbridge-grpc repository also contains generated
+programming language. The liftbridge-api repository also contains generated
 code for some languages which you can use in your client library (for example,
 the official Go client [relies on](https://github.com/liftbridge-io/go-liftbridge/blob/6fbf530bb220797fd91174d9f858fad3114dbc48/go.mod#L8)
-the generated Go code in liftbridge-grpc). If there is no generated code for
+the generated Go code in liftbridge-api). If there is no generated code for
 your language, feel free to make a PR adding it. Otherwise, you can simply
 vendor the generated code alongside your client library.
 
@@ -34,7 +34,9 @@ A high-level client has several operations:
   related NATS subjects if partitioned)
 - [`Subscribe`](#subscribe): creates an ephemeral subscription for a given stream that
   messages are received on
-- [`Publish`](#publish): publishes a new message to a NATS subject
+- [`Publish`](#publish): publishes a new message to a Liftbridge stream
+- [`PublishToSubject`](#publishtosubject): publishes a new message to a NATS
+  subject
 - [`FetchMetadata`](#fetchmetadata): retrieves metadata from the cluster
 - [`Close`](#close): closes any client connections to Liftbridge
 
@@ -57,18 +59,42 @@ type Client interface {
 	// Subscribe creates an ephemeral subscription for the given stream. It
 	// begins receiving messages starting at the configured position and waits
 	// for new messages when it reaches the end of the stream. The default
-	// start position is the end of the stream. It returns an ErrNoSuchStream
-	// if the given stream does not exist. Use a cancelable Context to close a
-	// subscription.
+	// start position is the end of the stream. It returns an
+	// ErrNoSuchPartition if the given stream or partition does not exist. Use
+	// a cancelable Context to close a subscription.
 	Subscribe(ctx context.Context, stream string, handler Handler, opts ...SubscriptionOption) error
 
-	// Publish publishes a new message to the NATS subject. If the AckPolicy is
-	// not NONE and a deadline is provided, this will synchronously block until
-	// the first ack is received. If the ack is not received in time, a
-	// DeadlineExceeded status code is returned. If an AckPolicy and deadline
-	// are configured, this returns the first Ack on success, otherwise it
-	// returns nil.
-	Publish(ctx context.Context, subject string, value []byte, opts ...MessageOption) (*proto.Ack, error)
+	// Publish publishes a new message to the Liftbridge stream. The partition
+	// that gets published to is determined by the provided partition or
+	// Partitioner passed through MessageOptions, if any. If a partition or
+	// Partitioner is not provided, this defaults to the base partition. This
+	// partition determines the underlying NATS subject that gets published to.
+	// To publish directly to a specific NATS subject, use the low-level
+	// PublishToSubject API.
+	//
+	// If the AckPolicy is not NONE and a deadline is provided, this will
+	// synchronously block until the ack is received. If the ack is not
+	// received in time, a DeadlineExceeded status code is returned. If an
+	// AckPolicy and deadline are configured, this returns the Ack on success,
+	// otherwise it returns nil.
+	Publish(ctx context.Context, stream string, value []byte, opts ...MessageOption) (Ack, error)
+
+	// PublishToSubject publishes a new message to the NATS subject. Note that
+	// because this publishes directly to a subject, there may be multiple (or
+	// no) streams that receive the message. As a result, MessageOptions
+	// related to partitioning will be ignored. To publish at the
+	// stream/partition level, use the high-level Publish API.
+	//
+	// If the AckPolicy is not NONE and a deadline is provided, this will
+	// synchronously block until the first ack is received. If an ack is not
+	// received in time, a DeadlineExceeded status code is returned. If an
+	// AckPolicy and deadline are configured, this returns the first Ack on
+	// success, otherwise it returns nil.
+	PublishToSubject(ctx context.Context, subject string, value []byte, opts ...MessageOption) (Ack, error)
+
+	// FetchMetadata returns cluster metadata including broker and stream
+	// information.
+	FetchMetadata(ctx context.Context) (*Metadata, error)
 }
 ```
 
@@ -79,7 +105,7 @@ type Client interface {
 // the NATS subject the stream is attached to, and name is the stream
 // identifier, unique per subject. It returns ErrStreamExists if a stream
 // with the given subject and name already exists.
-CreateStream(ctx context.Context, subject, name string, opts ...StreamOption) error
+func CreateStream(ctx context.Context, subject, name string, opts ...StreamOption) error
 ```
 
 `CreateStream` creates a new stream attached to a NATS subject or set of NATS
@@ -129,7 +155,7 @@ configure a stream. Supported options are:
 // start position is the end of the stream. It returns an ErrNoSuchStream
 // if the given stream does not exist. Use a cancelable Context to close a
 // subscription.
-Subscribe(ctx context.Context, stream string, handler Handler, opts ...SubscriptionOption) error
+func Subscribe(ctx context.Context, stream string, handler Handler, opts ...SubscriptionOption) error
 ```
 
 `Subscribe` is used to consume streams. By default, it begins reading messages
@@ -151,7 +177,7 @@ below. In other languages, this might be a stream mechanism instead of a
 callback.
 
 ```go
-type Handler func(msg *proto.Message, err error)
+type Handler func(msg Message, err error)
 ```
 
 The subscription options are the equivalent of optional named arguments used to
@@ -177,18 +203,25 @@ there will be functionality for consuming all partitions.
 ### Publish
 
 ```go
-// Publish publishes a new message to the NATS subject. If the AckPolicy is
-// not NONE and a deadline is provided, this will synchronously block until
-// the first ack is received. If the ack is not received in time, a
-// DeadlineExceeded status code is returned. If an AckPolicy and deadline
-// are configured, this returns the first Ack on success, otherwise it
-// returns nil.
-Publish(ctx context.Context, subject string, value []byte, opts ...MessageOption) (*proto.Ack, error)
+// Publish publishes a new message to the Liftbridge stream. The partition
+// that gets published to is determined by the provided partition or
+// Partitioner passed through MessageOptions, if any. If a partition or
+// Partitioner is not provided, this defaults to the base partition. This
+// partition determines the underlying NATS subject that gets published to.
+// To publish directly to a specific NATS subject, use the low-level
+// PublishToSubject API.
+//
+// If the AckPolicy is not NONE and a deadline is provided, this will
+// synchronously block until the ack is received. If the ack is not
+// received in time, a DeadlineExceeded status code is returned. If an
+// AckPolicy and deadline are configured, this returns the Ack on success,
+// otherwise it returns nil.
+func Publish(ctx context.Context, stream string, value []byte, opts ...MessageOption) (Ack, error)
 ```
 
-`Publish` sends a message to a NATS subject (and, in turn, any streams that
-match the subject). Since Liftbridge streams attach to normal NATS subjects,
-it's also possible to [publish messages directly to NATS](https://github.com/liftbridge-io/go-liftbridge#publishing-directly-with-nats)
+`Publish` sends a message to a Liftbridge stream. Since Liftbridge streams
+attach to normal NATS subjects, it's also possible to [publish messages
+directly to NATS](https://github.com/liftbridge-io/go-liftbridge#publishing-directly-with-nats)
 using a [NATS client](https://nats.io/download/). Liftbridge works fine with
 plain, opaque NATS messages, but it also extends NATS with a [protobuf-based
 envelope protocol](https://github.com/liftbridge-io/liftbridge-api). This
@@ -206,16 +239,17 @@ delivery. The default ack policy is `LEADER`, meaning the ack is sent once the
 partition leader has stored the message.
 
 `Publish` can send messages to particular stream partitions using a
-`Partitioner`. In effect, this publishes to a NATS subject derived from the one
-passed into `Publish` based on the partition, e.g. `foo`, `foo.1`, `foo.2`,
-etc. By default, it publishes to the base subject.
+`Partitioner` or an explicitly provided partition. By default, it publishes to
+the base partition. The partition determines the underlying NATS subject that
+gets published to. To publish directly to a specific NATS subject, use the
+low-level [`PublishToSubject`](#publishtosubject) API.
 
 In the Go client example above, `Publish` takes four arguments:
 
 | Argument | Type | Description | Required |
 |:----|:----|:----|:----|
 | context | context | A [context](https://golang.org/pkg/context/#Context) which is a Go idiom for passing things like a timeout, cancellation signal, and other values across API boundaries. For Liftbridge, this is primarily used for two things: request timeouts and cancellation. In other languages, this might be replaced by explicit arguments, optional named arguments, or other language-specific idioms. With `Publish`, the timeout is particularly important as it relates to acking. If an ack policy is set (see below) and a timeout is provided, `Publish` will block until the first ack is received. If the ack is not received in time, a timeout error is returned. If the ack policy is `NONE` or a timeout is not set, `Publish` returns as soon as the message has been published. | language-dependent |
-| subject | string | The NATS subject to publish to. | yes |
+| stream | string | The stream to publish to. | yes |
 | value | bytes | The message value to publish consisting of opaque bytes. | yes |
 | options | message options | Zero or more message options. These are used to pass in optional settings for the message. This is a common [Go pattern](https://dave.cheney.net/2014/10/17/functional-options-for-friendly-apis) for implementing extensible APIs. In other languages, this might be replaced with a builder pattern or optional named arguments. These `Publish` options are described below. | language-dependent |
 
@@ -245,19 +279,86 @@ described later. The Go interface definition is shown below.
 // Partitioner is used to map a message to a stream partition.
 type Partitioner interface {
 	// Partition computes the partition number for a given message.
-	Partition(msg *proto.Message, metadata *Metadata) int32
+	Partition(stream string, key, value []byte, metadata *Metadata) int32
 }
 ```
 
 [Implementation Guidance](#publish-implementation)
 
+### PublishToSubject
+
+```go
+// PublishToSubject publishes a new message to the NATS subject. Note that
+// because this publishes directly to a subject, there may be multiple (or
+// no) streams that receive the message. As a result, MessageOptions
+// related to partitioning will be ignored. To publish at the
+// stream/partition level, use the high-level Publish API.
+//
+// If the AckPolicy is not NONE and a deadline is provided, this will
+// synchronously block until the first ack is received. If an ack is not
+// received in time, a DeadlineExceeded status code is returned. If an
+// AckPolicy and deadline are configured, this returns the first Ack on
+// success, otherwise it returns nil.
+func PublishToSubject(ctx context.Context, subject string, value []byte, opts ...MessageOption) (Ack, error)
+```
+
+`PublishToSubject` sends a message to a NATS subject (and, in turn, any streams
+that match the subject). Since Liftbridge streams attach to normal NATS
+subjects, it's also possible to [publish messages directly to
+NATS](https://github.com/liftbridge-io/go-liftbridge#publishing-directly-with-nats)
+using a [NATS client](https://nats.io/download/). Liftbridge works fine with
+plain, opaque NATS messages, but it also extends NATS with a [protobuf-based
+envelope protocol](https://github.com/liftbridge-io/liftbridge-api). This
+allows publishers to add metadata to messages like a key, headers, and acking
+information. Liftbridge client libraries may provide helper methods to make it
+easy to create envelopes and deal with acks yourself using a NATS client
+directly ([described below](#low-level-publish-helpers)). However, the
+`PublishToSubject` API is intended to abstract this work away from you.
+
+`PublishToSubject` is a synchronous operation, meaning when it returns, the
+message has been successfully published. `PublishToSubject` can also be
+configured to block until a message acknowledgement (ack) is returned from the
+cluster. This is useful for ensuring a message has been stored and replicated,
+guaranteeing at-least-once delivery. The default ack policy is `LEADER`,
+meaning the ack is sent once the partition leader has stored the message.
+`PublishToSubject` only waits on the first ack received. This is important to
+be aware of since there may be multiple streams attached to a subject.
+
+In the Go client example above, `PublishToSubject` takes four arguments:
+
+| Argument | Type | Description | Required |
+|:----|:----|:----|:----|
+| context | context | A [context](https://golang.org/pkg/context/#Context) which is a Go idiom for passing things like a timeout, cancellation signal, and other values across API boundaries. For Liftbridge, this is primarily used for two things: request timeouts and cancellation. In other languages, this might be replaced by explicit arguments, optional named arguments, or other language-specific idioms. With `Publish`, the timeout is particularly important as it relates to acking. If an ack policy is set (see below) and a timeout is provided, `Publish` will block until the first ack is received. If the ack is not received in time, a timeout error is returned. If the ack policy is `NONE` or a timeout is not set, `Publish` returns as soon as the message has been published. | language-dependent |
+| subject | string | The NATS subject to publish to. | yes |
+| value | bytes | The message value to publish consisting of opaque bytes. | yes |
+| options | message options | Zero or more message options. These are used to pass in optional settings for the message. This is a common [Go pattern](https://dave.cheney.net/2014/10/17/functional-options-for-friendly-apis) for implementing extensible APIs. In other languages, this might be replaced with a builder pattern or optional named arguments. These `Publish` options are described below. | language-dependent |
+
+The publish message options are the equivalent of optional named arguments used
+to configure a message. Unlike `Publish`, `PublishToSubject` does not support
+message options relating to partitioning since it operates at a lower level.
+Supported options are:
+
+| Option | Type | Description | Default |
+|:----|:----|:----|:----|
+| AckInbox | string | Sets the NATS subject Liftbridge should publish the message ack to. If this is not set, the server will generate a random inbox. This generally does not need to be configured when using the `PublishToSubject` API since the server will handle acks for you. Instead, it's used if, for some reason, you wish to handle the ack yourself. | |
+| CorrelationID | string | Sets the identifier used to correlate an ack with the published message. If it's not set, the ack will not have a correlation ID. This generally does not need to be configured when using the `PublishToSubject` API since the server will handle acks for you. Instead, it's used if, for some reason, you wish to handle the ack yourself. | |
+| AckPolicyAll | bool | Sets the ack policy of the message to `ALL`. This means the ack will be sent when the message has been stored by all partition replicas. | false |
+| AckPolicyLeader | bool | Sets the ack policy of the message to `LEADER`. This means the ack will be sent when the partition leader has stored the message. This is the default ack policy if not otherwise set. | true |
+| AckPolicyNone | bool | Sets the ack policy of the message to `NONE`. This means no ack will be sent. | false |
+| Header | string and bytes | String and opaque bytes representing a key-value pair to set as a header on the message. This may overwrite previously set headers. Client libraries may choose to forgo this option in favor of the bulk `Headers` option below. This is a convenience for setting a single header. | |
+| Headers | map of strings to bytes | Map of strings to opaque bytes representing key-value pairs to set as headers on the message. This may overwrite previously set headers. | |
+| Key | bytes | Opaque bytes used for the message key. If Liftbridge has stream compaction enabled, the stream will retain only the last message for each key (if not set, the message is always retained). | |
+
+[Implementation Guidance](#publishtosubject-implementation)
+
 #### Low-Level Publish Helpers
 
 Because Liftbridge is simply a consumer of NATS, it's also possible to publish
-messages directly to NATS rather than using Liftbridge's `Publish` API. This
-can be useful for extending the capabilities of NATS in existing systems. This
-means we can publish NATS messages as normal, as shown in the example Go code
-below, and Liftbridge streams will be able to receive them.
+messages directly to NATS rather than using Liftbridge's `Publish` and
+`PublishToSubject` APIs. This can be useful for extending the capabilities of
+NATS in existing systems. This means we can publish NATS messages as normal, as
+shown in the example Go code below, and Liftbridge streams will be able to
+receive them.
 
 ```go
 package main
@@ -289,38 +390,38 @@ func NewMessage(value []byte, options ...MessageOption) []byte
 ```
 
 `NewMessage` creates a Liftbridge message envelope serialized to bytes ready
-for publishing to NATS. This consists of a four bytes ("LIFT"), which is
-referred to as the envelope cookie, followed by the serialized message
-protobuf. It takes the same arguments as `Publish` (see above) with the
-exception of the context and subject.
+for publishing to NATS. This consists of an [envelope
+header](envelope_protocol.md) followed by the serialized message protobuf. It
+takes the same arguments as `Publish` (see above) with the exception of the
+context and subject.
 
-Note that the envelope-cookie protocol does not need to be implemented in the
+Note that the envelope protocol does not need to be implemented in the
 `Publish` API since the envelope serialization is handled by the server.
 
 ##### UnmarshalMessage
 
 ```go
 // UnmarshalMessage deserializes a message from the given byte slice. It
-// returns a bool indicating if the given data was actually a Message or not.
-func UnmarshalMessage(data []byte) (*proto.Message, bool)
+// returns an error if the given data is not actually a Message.
+func UnmarshalMessage(data []byte) (Message, error)
 ```
 
 `UnmarshalMessage` is a helper method which effectively does the reverse of
-`NewMessage`, taking a serialized message and returning a deserialized message
-object or indication the data is not actually a message.
+`NewMessage`, taking a serialized message envelope and returning a deserialized
+message object or indication the data is not actually a message.
 
 ##### UnmarshalAck
 
 ```go
 // UnmarshalAck deserializes an Ack from the given byte slice. It returns an
 // error if the given data is not actually an Ack.
-func UnmarshalAck(data []byte) (*proto.Ack, error)
+func UnmarshalAck(data []byte) (Ack, error)
 ```
 
 `UnmarshalAck` is used to deserialize a message ack received on a NATS
 subscription. It takes a single argument consisting of the ack bytes as
 received from NATS and throws an error/exception if the bytes are not actually
-a serialized ack protobuf.
+a serialized ack envelope.
 
 This is useful for handling acks for messages published with a set ack inbox
 (i.e. the NATS subject Liftbridge will publish message acks to). For example:
@@ -333,7 +434,7 @@ nc, _ := nats.Connect(nats.DefaultURL)
 nc.Subscribe("my.ack.inbox", func(msg *nats.Msg) {
     ack, _ := lift.UnmarshalAck(msg.Data)
     // CorrelationId can be used to correlate an ack with a particular message.
-    if ack.CorrelationId == "123" {
+    if ack.CorrelationID() == "123" {
         println("message acked!")
     }
 })
@@ -389,7 +490,7 @@ In the Go client example above, `FetchMetadata` takes one argument:
 
 ```go
 // Close the client connection.
-Close() error
+func Close() error
 ```
 
 `Close` simply releases all resources associated with the client, namely all
@@ -422,6 +523,7 @@ idempotency reasons. This resilient RPC method can be used for RPCs such as:
 
 - `CreateStream`
 - `Publish`
+- `PublishToSubject`
 - `FetchMetadata`
 
 The Go implementation of this, called `doResilientRPC`, is shown below along
@@ -632,76 +734,83 @@ of behavior to point out with the dispatching component:
 `Publish` involves constructing a `Message` protobuf, determining the partition
 to publish to, and then making the publish request to the server using the
 [resilient RPC method](#rpcs) described above. The Go implementation of this is
-shown below, including a helper function `partition` which determines the
-partition ID to publish the message to.
+shown below, including two helper functions, `partition`, which determines the
+partition ID to publish the message to, and `publish`, which performs the
+actual RPC.
 
 ```go
-// Publish publishes a new message to the NATS subject. If the AckPolicy is not
-// NONE and a deadline is provided, this will synchronously block until the
-// first ack is received. If the ack is not received in time, a
-// DeadlineExceeded status code is returned. If an AckPolicy and deadline are
-// configured, this returns the first Ack on success, otherwise it returns nil.
-func (c *client) Publish(ctx context.Context, subject string, value []byte,
-	options ...MessageOption) (*proto.Ack, error) {
+// Publish publishes a new message to the Liftbridge stream. The partition that
+// gets published to is determined by the provided partition or Partitioner
+// passed through MessageOptions, if any. If a partition or Partitioner is not
+// provided, this defaults to the base partition. This partition determines the
+// underlying NATS subject that gets published to.  To publish directly to a
+// spedcific NATS subject, use the low-level PublishToSubject API.
+//
+// If the AckPolicy is not NONE and a deadline is provided, this will
+// synchronously block until the ack is received. If the ack is not received in
+// time, a DeadlineExceeded status code is returned. If an AckPolicy and
+// deadline are configured, this returns the Ack on success, otherwise it
+// returns nil.
+func (c *client) Publish(ctx context.Context, stream string, value []byte,
+	options ...MessageOption) (Ack, error) {
 
 	opts := &MessageOptions{Headers: make(map[string][]byte)}
 	for _, opt := range options {
 		opt(opts)
 	}
 
-	msg := &proto.Message{
-		Subject:       subject,
-		Key:           opts.Key,
-		Value:         value,
-		AckInbox:      opts.AckInbox,
-		CorrelationId: opts.CorrelationID,
-		AckPolicy:     opts.AckPolicy,
-	}
-
 	// Determine which partition to publish to.
-	partition, err := c.partition(ctx, msg, opts)
+	partition, err := c.partition(ctx, stream, opts.Key, value, opts)
 	if err != nil {
 		return nil, err
 	}
 
-	// Publish to the appropriate partition subject.
-	if partition > 0 {
-		msg.Subject = fmt.Sprintf("%s.%d", msg.Subject, partition)
+	req := &proto.PublishRequest{
+		Stream:        stream,
+		Partition:     partition,
+		Key:           opts.Key,
+		Value:         value,
+		AckInbox:      opts.AckInbox,
+		CorrelationId: opts.CorrelationID,
+		AckPolicy:     opts.AckPolicy.toProto(),
 	}
 
-	var (
-		req = &proto.PublishRequest{Message: msg}
-		ack *proto.Ack
-	)
-	err = c.doResilientRPC(func(client proto.APIClient) error {
-		resp, err := client.Publish(ctx, req)
-		if err == nil {
-			ack = resp.Ack
-		}
-		return err
-	})
-	return ack, err
+	return c.publish(ctx, req)
 }
 
 // partition determines the partition ID to publish the message to. If a
 // partition was explicitly provided, it will be returned. If a Partitioner was
 // provided, it will be used to compute the partition. Otherwise, 0 will be
 // returned.
-func (c *client) partition(ctx context.Context, msg *proto.Message, opts *MessageOptions) (int32, error) {
+func (c *client) partition(ctx context.Context, stream string, key, value []byte,
+	opts *MessageOptions) (int32, error) {
+
 	var partition int32
 	// If a partition is explicitly provided, use it.
 	if opts.Partition != nil {
 		partition = *opts.Partition
 	} else if opts.Partitioner != nil {
 		// Make sure we have metadata for the stream and, if not, update it.
-		metadata, err := c.waitForSubjectMetadata(ctx, msg.Subject)
+		metadata, err := c.waitForStreamMetadata(ctx, stream)
 		if err != nil {
 			return 0, err
 		}
-		partition = opts.Partitioner.Partition(msg, metadata)
+		partition = opts.Partitioner.Partition(stream, key, value, metadata)
 	}
 	return partition, nil
 }
+
+func (c *client) publish(ctx context.Context, req *proto.PublishRequest) (Ack, error) {
+	var ack *proto.Ack
+	err := c.doResilientRPC(func(client proto.APIClient) error {
+		resp, err := client.Publish(ctx, req)
+		if err == nil {
+			ack = resp.Ack
+		}
+		return err
+	})
+	return newProtoAck(ack), err
+}
 ```
 
 The partition to publish to is determined in the following way:
@@ -711,17 +820,6 @@ The partition to publish to is determined in the following way:
 2. If a `Partitioner` is provided, use it to compute the partition.
 3. If neither of the above apply, use partition 0 (the default partition).
 
-The partition is used to derive the actual NATS subject the message is
-published to. The table below shows an example of this partition-subject
-mapping logic.
-
-| Base Subject | Partition | Derived Subject |
-|:----|:----|:----|
-| foo | 0 | foo |
-| foo | 1 | foo.1 |
-| foo | 2 | foo.2 |
-| foo | 3 | foo.3 |
-
 Note that `Publish` is a synchronous operation. After the RPC returns, the
 message has been published. The server handles message acks, if applicable. If
 a publish is expecting an ack, the RPC will block until the ack is received or
@@ -745,18 +843,13 @@ var hasher = crc32.ChecksumIEEE
 type keyPartitioner struct{}
 
 // Partition computes the partition number for a given message by hashing the
-// key and modding by the number of partitions for the first stream found with
-// the subject of the message. This does not work with streams containing
-// wildcards in their subjects, e.g. "foo.*", since this matches on the subject
-// literal of the published message. This also has undefined behavior if there
-// are multiple streams for the given subject.
-func (k *keyPartitioner) Partition(msg *proto.Message, metadata *Metadata) int32 {
-	key := msg.Key
+// key and modding by the number of stream partitions.
+func (k *keyPartitioner) Partition(stream string, key, value []byte, metadata *Metadata) int32 {
 	if key == nil {
 		key = []byte("")
 	}
 
-	partitions := getPartitionCount(msg.Subject, metadata)
+	partitions := metadata.PartitionCountForStream(stream)
 	if partitions == 0 {
 		return 0
 	}
@@ -777,22 +870,18 @@ type roundRobinPartitioner struct {
 }
 
 // Partition computes the partition number for a given message in a round-robin
-// fashion by atomically incrementing a counter for the message subject and
-// modding by the number of partitions for the first stream found with the
-// subject. This does not work with streams containing wildcards in their
-// subjects, e.g. "foo.*", since this matches on the subject literal of the
-// published message. This also has undefined behavior if there are multiple
-// streams for the given subject.
-func (r *roundRobinPartitioner) Partition(msg *proto.Message, metadata *Metadata) int32 {
-	partitions := getPartitionCount(msg.Subject, metadata)
+// fashion by atomically incrementing a counter for the message stream and
+// modding by the number of stream partitions.
+func (r *roundRobinPartitioner) Partition(stream string, key, value []byte, metadata *Metadata) int32 {
+	partitions := metadata.PartitionCountForStream(stream)
 	if partitions == 0 {
 		return 0
 	}
 	r.Lock()
-	counter, ok := r.subjectCounterMap[msg.Subject]
+	counter, ok := r.streamCounterMap[stream]
 	if !ok {
-		counter = new(subjectCounter)
-		r.subjectCounterMap[msg.Subject] = counter
+		counter = new(streamCounter)
+		r.streamCounterMap[stream] = counter
 	}
 	r.Unlock()
 	counter.Lock()
@@ -801,16 +890,46 @@ func (r *roundRobinPartitioner) Partition(msg *proto.Message, metadata *Metadata
 	counter.Unlock()
 	return count % partitions
 }
+```
+
+### PublishToSubject Implementation
+
+`PublishToSubject` implementation is similar to that of
+[`Publish`](#publish-implementation), but rather than sending a stream and
+partition in the `PublishRequest`, it only sends a subject. It can reuse the
+same publish helper used by `Publish`. The Go implementations of this is shown
+below:
+
+```go
+// PublishToSubject publishes a new message to the NATS subject. Note that
+// because this publishes directly to a subject, there may be multiple (or no)
+// streams that receive the message. As a result, MessageOptions related to
+// partitioning will be ignored. To publish at the stream/partition level, use
+// the high-level Publish API.
+//
+// If the AckPolicy is not NONE and a deadline is provided, this will
+// synchronously block until the first ack is received. If an ack is not
+// received in time, a DeadlineExceeded status code is returned. If an
+// AckPolicy and deadline are configured, this returns the first Ack on
+// success, otherwise it returns nil.
+func (c *client) PublishToSubject(ctx context.Context, subject string, value []byte,
+	options ...MessageOption) (Ack, error) {
 
-func getPartitionCount(subject string, metadata *Metadata) int32 {
-	counts := metadata.PartitionCountsForSubject(subject)
+	opts := &MessageOptions{Headers: make(map[string][]byte)}
+	for _, opt := range options {
+		opt(opts)
+	}
 
-	// Get the first matching stream's count.
-	for _, count := range counts {
-		return count
+	req := &proto.PublishRequest{
+		Subject:       subject,
+		Key:           opts.Key,
+		Value:         value,
+		AckInbox:      opts.AckInbox,
+		CorrelationId: opts.CorrelationID,
+		AckPolicy:     opts.AckPolicy.toProto(),
 	}
 
-	return 0
+	return c.publish(ctx, req)
 }
 ```
 
@@ -840,10 +959,9 @@ func (m *Metadata) GetStream(name string) *StreamInfo {}
 // does not match on wildcard subjects, e.g.  "foo.*".
 func (m *Metadata) GetStreams(subject string) []*StreamInfo {}
 
-// PartitionCountsForSubject returns a map containing stream names and the
-// number of partitions for the stream. This does not match on wildcard
-// subjects, e.g. "foo.*".
-func (m *Metadata) PartitionCountsForSubject(subject string) map[string]int32 {}
+// PartitionCountForStream returns the number of partitions for the given
+// stream.
+func (m *Metadata) PartitionCountForStream(stream string) int32 {}
 ```
 
 `BrokerInfo` is an immutable object containing server information like ID,
diff --git a/documentation/envelope_protocol.md b/documentation/envelope_protocol.md
new file mode 100644
index 0000000..178ba50
--- /dev/null
+++ b/documentation/envelope_protocol.md
@@ -0,0 +1,67 @@
+---
+id: envelope-protocol
+title: Envelope Protocol
+---
+
+Liftbridge works by wrapping NATS with higher-level stream APIs. It is also
+possible for a client to publish messages to Liftbridge via NATS directly.
+Liftbridge accepts plain NATS messages, allowing it to make existing subjects
+durable without any publisher changes. However, these messages will not have
+features such as acks.
+
+In order to opt into Liftbridge-specific features, the message must be prefixed
+with an envelope header and be encoded as a `Message` (defined in
+[api.proto](https://github.com/liftbridge-io/liftbridge-api/blob/master/api.proto)).
+Stream acks sent over NATS use the same envelope protocol. This envelope
+protocol is described below. Refer
+[here](https://github.com/liftbridge-io/liftbridge-api) for more information on
+the Liftbridge API.
+
+## Liftbridge Envelope Header
+
+```plaintext
+0               8               16              24              32
+├───────────────┴───────────────┴───────────────┴───────────────┤
+│                          Magic Number                         │
+├───────────────┬───────────────┬───────────────┬───────────────┤
+│    Version    │   HeaderLen   │     Flags     │    Reserved   │
+├───────────────┴───────────────┴───────────────┴───────────────┤
+│                       CRC-32C (optional)                      │
+└───────────────────────────────────────────────────────────────┘
+```
+
+### Magic Number [4 bytes]
+
+The Liftbridge magic number is `B9 0E 43 B4`. This was chosen by random but
+deliberately restricted to invalid UTF-8 to reduce the chance of a collision.
+This was also verified to not match known file signatures.
+
+### Version [1 byte]
+
+The version byte allows for future protocol upgrades. This should only be
+bumped if the envelope format changes or if the message encoding changes in a
+non-backwards-compatible way. Adding fields to the messages should not require
+a version bump.
+
+### HeaderLen [1 byte]
+
+The header length is the offset of the payload. This is included primarily for
+safety.
+
+### Flags [1 byte]
+
+The flag bits are defined as follows:
+
+| Bit | Description     |
+| --- | --------------- |
+| 0   | CRC-32C enabled |
+
+### Reserved [1 byte]
+
+Reserved for future use.
+
+### CRC-32C [4 bytes, optional]
+
+The CRC-32C (Castagnoli) is the checksum of the payload (i.e. from HeaderLen to
+the end). This is optional but should significantly reduce the chance that a
+random NATS message is interpreted as a Liftbridge message.
diff --git a/go.mod b/go.mod
index d443c57..d4b4df5 100644
--- a/go.mod
+++ b/go.mod
@@ -5,12 +5,12 @@ go 1.13
 require (
 	github.com/Workiva/go-datastructures v1.0.50
 	github.com/dustin/go-humanize v1.0.0
-	github.com/golang/protobuf v1.3.2
+	github.com/golang/protobuf v1.3.3
 	github.com/hako/durafmt v0.0.0-20191009132224-3f39dc1ed9f4
 	github.com/hashicorp/raft v1.1.1
 	github.com/hashicorp/raft-boltdb v0.0.0-20191021154308-4207f1bf0617
-	github.com/liftbridge-io/go-liftbridge v0.0.0-20191106180712-4f7e4b9d8611
-	github.com/liftbridge-io/liftbridge-api v0.0.0-20190910222614-5694b15f251d
+	github.com/liftbridge-io/go-liftbridge v1.0.0-alpha
+	github.com/liftbridge-io/liftbridge-api v1.0.0-alpha
 	github.com/liftbridge-io/nats-on-a-log v0.0.0-20190703144237-760cefbfc85e
 	github.com/natefinch/atomic v0.0.0-20150920032501-a62ce929ffcc
 	github.com/nats-io/nats-server/v2 v2.1.0
@@ -21,5 +21,5 @@ require (
 	github.com/sirupsen/logrus v1.4.2
 	github.com/stretchr/testify v1.4.0
 	github.com/urfave/cli v1.22.1
-	google.golang.org/grpc v1.25.0
+	google.golang.org/grpc v1.27.1
 )
diff --git a/go.sum b/go.sum
index efcfe43..eb8789a 100644
--- a/go.sum
+++ b/go.sum
@@ -31,6 +31,7 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
+github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
@@ -50,6 +51,8 @@ github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5y
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.3 h1:gyjaxf+svBWX08ZjK86iN9geUJF0H6gp2IRKX6Nf6/I=
+github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
 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=
@@ -112,13 +115,24 @@ github.com/liftbridge-io/go-liftbridge v0.0.0-20190831233708-6fbf530bb220/go.mod
 github.com/liftbridge-io/go-liftbridge v0.0.0-20191106171334-84163faf9fdd/go.mod h1:Q1PGCu9GUniz/0iHucxMwg2tiHwqhg8JrmIFOaPD88s=
 github.com/liftbridge-io/go-liftbridge v0.0.0-20191106180712-4f7e4b9d8611 h1:VB8/4FRlmEaZ1toJC854gS+klexnEpFVghHAUN227Tk=
 github.com/liftbridge-io/go-liftbridge v0.0.0-20191106180712-4f7e4b9d8611/go.mod h1:E+5gzmn2VHTfwQ21r0bujaQrUe5o1UtHzCAr+Xpo47A=
+github.com/liftbridge-io/go-liftbridge v0.0.1-alpha.0.20200120163431-bfb99c041e6f h1:p8tufXYhYE0KLL5/ePtGCr05bUWxnfzb7Jl2bclGc4M=
+github.com/liftbridge-io/go-liftbridge v0.0.1-alpha.0.20200120163431-bfb99c041e6f/go.mod h1:UO2x6evHiWpw37/jEvVbKr4dzg17JHJB8f3dzBtMhow=
+github.com/liftbridge-io/go-liftbridge v1.0.0-alpha h1:Mgzb8ACTNDXAfSMGxbsU7Mg+maAZIDUgB1NE5rtGdPc=
+github.com/liftbridge-io/go-liftbridge v1.0.0-alpha/go.mod h1:2H2qD8RQzUwR6yOSM77OLsMAwn9XDNIlkqHWtJP6jnk=
 github.com/liftbridge-io/liftbridge v0.0.0-20190628061900-5f565727d49f/go.mod h1:DgUnOkzaVLgIOQ3W7ztMm0+xQmDaEM+tcgGCAdlZj8s=
 github.com/liftbridge-io/liftbridge v0.0.0-20190704001405-928a9d17c609/go.mod h1:d9iQ/6VLp7paFMnNukU2nV1PnNjkz2vpNQzYS4zIy8Y=
 github.com/liftbridge-io/liftbridge v0.0.0-20190831212313-ad1b5f9c2b17/go.mod h1:hnwhdV70dIFgIKvzBJt9rOfce19kE2a3ZGaRgSQmNOs=
 github.com/liftbridge-io/liftbridge v0.0.0-20191101170542-6684dd2aa80f/go.mod h1:l/Y7oCEKS+evzr19nCjeatDJU7SHwtjh6uIYGULUJfE=
 github.com/liftbridge-io/liftbridge v0.0.0-20191106173932-5d89c897d5b0/go.mod h1:Ysh4YXyf7Fvdeid4QJNmyYqDRWwwMAXNmhV3IPb1l3M=
+github.com/liftbridge-io/liftbridge v0.0.0-20200118200014-2e716da78b8a/go.mod h1:j3eqt63LZefajwMwcLeYtS/BEFHqPXA7ogvjJCsB1h8=
 github.com/liftbridge-io/liftbridge-api v0.0.0-20190910222614-5694b15f251d h1:yuAMJgvh7C/QPVAui2i+MF8X3wUzRPgdh/t0P9Gx/p4=
 github.com/liftbridge-io/liftbridge-api v0.0.0-20190910222614-5694b15f251d/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
+github.com/liftbridge-io/liftbridge-api v0.0.0-20200118013732-a474da01bc36 h1:O07A30xXqJdx6DKI+OQYB9OILob2nRfTde0Ok2VAzjs=
+github.com/liftbridge-io/liftbridge-api v0.0.0-20200118013732-a474da01bc36/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
+github.com/liftbridge-io/liftbridge-api v0.0.0-20200118015119-3db283f59b10 h1:k10Bg5EQTc/B5dTSU8cPW4Y0OkBlnMFuFGZwRmrPvJ4=
+github.com/liftbridge-io/liftbridge-api v0.0.0-20200118015119-3db283f59b10/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
+github.com/liftbridge-io/liftbridge-api v1.0.0-alpha h1:NwtHTgIdYNQXkLdmLHm9rguNvbtT4OHAs6FSSrHTq1Q=
+github.com/liftbridge-io/liftbridge-api v1.0.0-alpha/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
 github.com/liftbridge-io/liftbridge-grpc v0.0.0-20190829220806-66e3ee4b7943/go.mod h1:ObGO38WdO4ldLsa2oUFcultUk0rggc+yZWcBb7qjnDI=
 github.com/liftbridge-io/liftbridge-grpc v0.0.0-20190910222614-5694b15f251d/go.mod h1:ObGO38WdO4ldLsa2oUFcultUk0rggc+yZWcBb7qjnDI=
 github.com/liftbridge-io/nats-on-a-log v0.0.0-20180718011723-80d0727461af/go.mod h1:4tC6R+N3facyfCwDuuuLkFF/25ceiZEwoQUzIez2dVo=
@@ -145,6 +159,8 @@ github.com/nats-io/jwt v0.2.6/go.mod h1:mQxQ0uHQ9FhEVPIcTSKwx2lqZEpXWWcCgA7R6NrW
 github.com/nats-io/jwt v0.2.8/go.mod h1:mQxQ0uHQ9FhEVPIcTSKwx2lqZEpXWWcCgA7R6NrWvvY=
 github.com/nats-io/jwt v0.3.0 h1:xdnzwFETV++jNc4W1mw//qFyJGb2ABOombmZJQS4+Qo=
 github.com/nats-io/jwt v0.3.0/go.mod h1:fRYCDE99xlTsqUzISS1Bi75UBJ6ljOJQOAAu5VglpSg=
+github.com/nats-io/jwt v0.3.2 h1:+RB5hMpXUUA2dfxuhBTEkMOrYmM+gKIZYS1KjSostMI=
+github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU=
 github.com/nats-io/nats-server v1.4.1 h1:Ul1oSOGNV/L8kjr4v6l2f9Yet6WY+LevH1/7cRZ/qyA=
 github.com/nats-io/nats-server v1.4.1/go.mod h1:c8f/fHd2B6Hgms3LtCaI7y6pC4WD1f4SUxcCud5vhBc=
 github.com/nats-io/nats-server/v2 v2.0.0/go.mod h1:RyVdsHHvY4B6c9pWG+uRLpZ0h0XsqiuKp2XCTurP5LI=
@@ -157,6 +173,8 @@ github.com/nats-io/nkeys v0.0.2/go.mod h1:dab7URMsZm6Z/jp9Z5UGa87Uutgc2mVpXLC4B7
 github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.2 h1:kWUa9fUdAOW426qaAD1V4+suF4On2eTr2n/FE/XSGpg=
 github.com/nats-io/nkeys v0.1.2/go.mod h1:oyD2oRRVULOkPJxjJ8mX1QY0ev0FNiCi7EoKNN09TYU=
+github.com/nats-io/nkeys v0.1.3 h1:6JrEfig+HzTH85yxzhSVbjHRJv9cn0p6n3IngIcM5/k=
+github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nuid v1.0.0/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
@@ -210,6 +228,8 @@ golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191105034135-c7e5f84aec59 h1:PyXRxSVbvzDGuqYXjHndV7xDzJ7w2K8KD9Ef8GB7KOE=
 golang.org/x/crypto v0.0.0-20191105034135-c7e5f84aec59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200214034016-1d94cc7ab1c6 h1:Sy5bstxEqwwbYs6n0/pBuxKENqOeZUgD45Gp3Q3pqLg=
+golang.org/x/crypto v0.0.0-20200214034016-1d94cc7ab1c6/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -242,6 +262,10 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191105084925-a882066a44e0 h1:QPlSTtPE2k6PZPasQUbzuK3p9JbS+vMXYVto8g/yrsg=
 golang.org/x/net v0.0.0-20191105084925-a882066a44e0/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa h1:F+8P+gmewFQYRk6JoLQLwjBCTu3mcIURZfNkVweuRKA=
+golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200219183655-46282727080f h1:dB42wwhNuwPvh8f+5zZWNcU+F2Xs/B9wXXwvUCOH7r8=
+golang.org/x/net v0.0.0-20200219183655-46282727080f/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 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=
@@ -270,6 +294,10 @@ golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191105231009-c1f44814a5cd h1:3x5uuvBgE6oaXJjCOvpCC1IpgJogqQ+PqGGU3ZxAgII=
 golang.org/x/sys v0.0.0-20191105231009-c1f44814a5cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200117145432-59e60aa80a0c h1:gUYreENmqtjZb2brVfUas1sC6UivSY8XwKwPo8tloLs=
+golang.org/x/sys v0.0.0-20200117145432-59e60aa80a0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200219091948-cb0a6d8edb6c h1:jceGD5YNJGgGMkJz79agzOln1K9TaZUjv5ird16qniQ=
+golang.org/x/sys v0.0.0-20200219091948-cb0a6d8edb6c/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 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
@@ -306,6 +334,10 @@ google.golang.org/genproto v0.0.0-20190701230453-710ae3a149df/go.mod h1:z3L6/3dT
 google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
 google.golang.org/genproto v0.0.0-20191028173616-919d9bdd9fe6 h1:UXl+Zk3jqqcbEVV7ace5lrt4YdA4tXiz3f/KbmD29Vo=
 google.golang.org/genproto v0.0.0-20191028173616-919d9bdd9fe6/go.mod h1:n3cpQtvxv34hfy77yVDNjmbRyujviMdxYliBSkLhpCc=
+google.golang.org/genproto v0.0.0-20200117163144-32f20d992d24 h1:wDju+RU97qa0FZT0QnZDg9Uc2dH0Ql513kFvHocz+WM=
+google.golang.org/genproto v0.0.0-20200117163144-32f20d992d24/go.mod h1:n3cpQtvxv34hfy77yVDNjmbRyujviMdxYliBSkLhpCc=
+google.golang.org/genproto v0.0.0-20200218151345-dad8c97a84f5 h1:jB9+PJSvu5tBfmJHy/OVapFdjDF3WvpkqRhxqrmzoEU=
+google.golang.org/genproto v0.0.0-20200218151345-dad8c97a84f5/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/grpc v1.16.0/go.mod h1:0JHn/cJsOMiMfNA9+DeHDlAU7KAAB5GDlYFpa9MZMio=
 google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
@@ -316,6 +348,11 @@ google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyac
 google.golang.org/grpc v1.24.0/go.mod h1:XDChyiUovWa60DnaeDeZmSW86xtLtjtZbwvSiRnRtcA=
 google.golang.org/grpc v1.25.0 h1:ItERT+UbGdX+s4u+nQNlVM/Q7cbmf7icKfvzbWqVtq0=
 google.golang.org/grpc v1.25.0/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
+google.golang.org/grpc v1.26.0 h1:2dTRdpdFEEhJYQD8EMLB61nnrzSCTbG38PhqdhvOltg=
+google.golang.org/grpc v1.26.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
+google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
+google.golang.org/grpc v1.27.1 h1:zvIju4sqAGvwKspUQOhwnpcqSbzi7/H6QomNNjTL4sk=
+google.golang.org/grpc v1.27.1/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/server/api.go b/server/api.go
index 7c76558..f902621 100644
--- a/server/api.go
+++ b/server/api.go
@@ -143,34 +143,42 @@ func (a *apiServer) FetchMetadata(ctx context.Context, req *client.FetchMetadata
 // is returned.
 func (a *apiServer) Publish(ctx context.Context, req *client.PublishRequest) (
 	*client.PublishResponse, error) {
-	if req.Message == nil {
-		a.logger.Errorf("api: Failed to publish message: message is nil")
-		return nil, status.Error(codes.InvalidArgument, "Message is nil")
+	subject, err := a.getPublishSubject(req)
+	if err != nil {
+		return nil, err
+	}
+	a.logger.Debugf("api: Publish [subject=%s]", subject)
+
+	if req.AckInbox == "" {
+		req.AckInbox = nuid.Next()
 	}
-	a.logger.Debugf("api: Publish [subject=%s]", req.Message.Subject)
 
-	if req.Message.AckInbox == "" {
-		req.Message.AckInbox = nuid.Next()
+	msg := &client.Message{
+		Key:           req.Key,
+		Value:         req.Value,
+		Stream:        req.Stream,
+		Subject:       subject,
+		ReplySubject:  req.ReplySubject,
+		Headers:       req.Headers,
+		AckInbox:      req.AckInbox,
+		CorrelationId: req.CorrelationId,
+		AckPolicy:     req.AckPolicy,
 	}
 
-	msg, err := req.Message.Marshal()
+	buf, err := proto.MarshalEnvelope(msg)
 	if err != nil {
-		a.logger.Errorf("api: Failed to publish message: %v", err.Error())
+		a.logger.Errorf("api: Failed to marshal message: %v", err.Error())
 		return nil, err
 	}
 
-	buf := make([]byte, envelopeCookieLen+len(msg))
-	copy(buf[0:], envelopeCookie)
-	copy(buf[envelopeCookieLen:], msg)
-
 	// If AckPolicy is NONE or a timeout isn't specified, then we will fire and
 	// forget.
 	var (
 		resp           = new(client.PublishResponse)
 		_, hasDeadline = ctx.Deadline()
 	)
-	if req.Message.AckPolicy == client.AckPolicy_NONE || !hasDeadline {
-		if err := a.ncPublishes.Publish(req.Message.Subject, buf); err != nil {
+	if req.AckPolicy == client.AckPolicy_NONE || !hasDeadline {
+		if err := a.ncPublishes.Publish(subject, buf); err != nil {
 			a.logger.Errorf("api: Failed to publish message: %v", err)
 			return nil, err
 		}
@@ -178,10 +186,28 @@ func (a *apiServer) Publish(ctx context.Context, req *client.PublishRequest) (
 	}
 
 	// Otherwise we need to publish and wait for the ack.
-	resp.Ack, err = a.publishSync(ctx, req.Message.Subject, req.Message.AckInbox, buf)
+	resp.Ack, err = a.publishSync(ctx, subject, req.AckInbox, buf)
 	return resp, err
 }
 
+func (a *apiServer) getPublishSubject(req *client.PublishRequest) (string, error) {
+	if req.Subject != "" {
+		return req.Subject, nil
+	}
+	if req.Stream == "" {
+		return "", status.Error(codes.InvalidArgument, "No stream or subject provided")
+	}
+	stream := a.metadata.GetStream(req.Stream)
+	if stream == nil {
+		return "", status.Error(codes.NotFound, fmt.Sprintf("No such stream: %s", req.Stream))
+	}
+	subject := stream.subject
+	if req.Partition > 0 {
+		subject = fmt.Sprintf("%s.%d", subject, req.Partition)
+	}
+	return subject, nil
+}
+
 func (a *apiServer) publishSync(ctx context.Context, subject,
 	ackInbox string, msg []byte) (*client.Ack, error) {
 
@@ -212,7 +238,7 @@ func (a *apiServer) publishSync(ctx context.Context, subject,
 	}
 
 	ack := new(client.Ack)
-	if err := ack.Unmarshal(ackMsg.Data); err != nil {
+	if err := proto.UnmarshalEnvelope(ackMsg.Data, ack); err != nil {
 		a.logger.Errorf("api: Invalid ack for publish: %v", err)
 		return nil, err
 	}
@@ -257,13 +283,15 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 			headers := m.Headers()
 			var (
 				msg = &client.Message{
-					Offset:    offset,
-					Key:       m.Key(),
-					Value:     m.Value(),
-					Timestamp: timestamp,
-					Headers:   headers,
-					Subject:   string(headers["subject"]),
-					Reply:     string(headers["reply"]),
+					Stream:       partition.Stream,
+					Partition:    partition.Id,
+					Offset:       offset,
+					Key:          m.Key(),
+					Value:        m.Value(),
+					Timestamp:    timestamp,
+					Headers:      headers,
+					Subject:      string(headers["subject"]),
+					ReplySubject: string(headers["reply"]),
 				}
 			)
 			select {
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index e17c11a..8411d1e 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -17,7 +17,6 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/liftbridge-io/liftbridge/server/logger"
-	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 // ErrSegmentNotFound is returned if the segment could not be found.
@@ -212,7 +211,7 @@ func (l *commitLog) open() error {
 
 // Append writes the given batch of messages to the log and returns their
 // corresponding offsets in the log.
-func (l *commitLog) Append(msgs []*proto.Message) ([]int64, error) {
+func (l *commitLog) Append(msgs []*Message) ([]int64, error) {
 	if _, err := l.checkAndPerformSplit(); err != nil {
 		return nil, err
 	}
diff --git a/server/proto/encoder.go b/server/commitlog/encoder.go
similarity index 63%
rename from server/proto/encoder.go
rename to server/commitlog/encoder.go
index 7340a52..89b0574 100644
--- a/server/proto/encoder.go
+++ b/server/commitlog/encoder.go
@@ -1,4 +1,4 @@
-package proto
+package commitlog
 
 import (
 	"encoding/binary"
@@ -7,16 +7,16 @@ import (
 )
 
 var (
-	// Encoding is the byte order to use for serialization.
-	Encoding = binary.BigEndian
+	// encoding is the byte order to use for internal disk serialization.
+	encoding = binary.BigEndian
 
 	errInvalidStringLength    = errors.New("invalid string length")
 	errInvalidArrayLength     = errors.New("invalid array length")
 	errInvalidByteSliceLength = errors.New("invalid byteslice length")
 )
 
-// PacketEncoder is used to serialize an object.
-type PacketEncoder interface {
+// packetEncoder is used to serialize an object.
+type packetEncoder interface {
 	PutBool(in bool)
 	PutInt8(in int8)
 	PutInt16(in int16)
@@ -30,33 +30,33 @@ type PacketEncoder interface {
 	PutStringArray(in []string) error
 	PutInt32Array(in []int32) error
 	PutInt64Array(in []int64) error
-	Push(pe PushEncoder)
+	Push(pe pushEncoder)
 	Pop()
 }
 
-// PushEncoder is used to push an operation onto the stack to perform later
+// pushEncoder is used to push an operation onto the stack to perform later
 // once serialized bytes are filled.
-type PushEncoder interface {
+type pushEncoder interface {
 	SaveOffset(in int)
 	ReserveSize() int
 	Fill(curOffset int, buf []byte) error
 }
 
-// Encoder is a struct that can be serialized.
-type Encoder interface {
-	Encode(e PacketEncoder) error
+// encoder is a struct that can be serialized.
+type encoder interface {
+	Encode(e packetEncoder) error
 }
 
-// Encode serializes the struct to bytes.
-func Encode(e Encoder) ([]byte, error) {
-	lenEnc := new(LenEncoder)
+// encode serializes the struct to bytes.
+func encode(e encoder) ([]byte, error) {
+	lenEnc := new(lenEncoder)
 	err := e.Encode(lenEnc)
 	if err != nil {
 		return nil, err
 	}
 
 	b := make([]byte, lenEnc.Length)
-	byteEnc := NewByteEncoder(b)
+	byteEnc := newByteEncoder(b)
 	err = e.Encode(byteEnc)
 	if err != nil {
 		return nil, err
@@ -65,39 +65,39 @@ func Encode(e Encoder) ([]byte, error) {
 	return b, nil
 }
 
-// LenEncoder is a PacketEncoder that tracks the running length of serialized
+// lenEncoder is a packetEncoder that tracks the running length of serialized
 // bytes.
-type LenEncoder struct {
+type lenEncoder struct {
 	Length int
 }
 
 // PutBool increments length for a bool.
-func (e *LenEncoder) PutBool(in bool) {
+func (e *lenEncoder) PutBool(in bool) {
 	e.Length++
 }
 
 // PutInt8 increments length for an int8.
-func (e *LenEncoder) PutInt8(in int8) {
+func (e *lenEncoder) PutInt8(in int8) {
 	e.Length++
 }
 
 // PutInt16 increments length for an int16.
-func (e *LenEncoder) PutInt16(in int16) {
+func (e *lenEncoder) PutInt16(in int16) {
 	e.Length += 2
 }
 
 // PutInt32 increments length for an int32.
-func (e *LenEncoder) PutInt32(in int32) {
+func (e *lenEncoder) PutInt32(in int32) {
 	e.Length += 4
 }
 
 // PutInt64 increments length for an int64.
-func (e *LenEncoder) PutInt64(in int64) {
+func (e *lenEncoder) PutInt64(in int64) {
 	e.Length += 8
 }
 
 // PutArrayLength increments length for an array size.
-func (e *LenEncoder) PutArrayLength(in int) error {
+func (e *lenEncoder) PutArrayLength(in int) error {
 	if in > math.MaxInt32 {
 		return errInvalidArrayLength
 	}
@@ -108,7 +108,7 @@ func (e *LenEncoder) PutArrayLength(in int) error {
 // arrays
 
 // PutBytes increments length for a size-prefixed byte array.
-func (e *LenEncoder) PutBytes(in []byte) error {
+func (e *lenEncoder) PutBytes(in []byte) error {
 	e.Length += 4
 	if in == nil {
 		return nil
@@ -121,7 +121,7 @@ func (e *LenEncoder) PutBytes(in []byte) error {
 }
 
 // PutRawBytes increments length for a raw byte array.
-func (e *LenEncoder) PutRawBytes(in []byte) error {
+func (e *lenEncoder) PutRawBytes(in []byte) error {
 	if len(in) > math.MaxInt32 {
 		return errInvalidByteSliceLength
 	}
@@ -130,7 +130,7 @@ func (e *LenEncoder) PutRawBytes(in []byte) error {
 }
 
 // PutString increments length for a string.
-func (e *LenEncoder) PutString(in string) error {
+func (e *lenEncoder) PutString(in string) error {
 	e.Length += 2
 	if len(in) > math.MaxInt16 {
 		return errInvalidStringLength
@@ -140,7 +140,7 @@ func (e *LenEncoder) PutString(in string) error {
 }
 
 // PutNullableString increments length for a nullable string.
-func (e *LenEncoder) PutNullableString(in *string) error {
+func (e *lenEncoder) PutNullableString(in *string) error {
 	if in == nil {
 		e.Length += 2
 		return nil
@@ -149,7 +149,7 @@ func (e *LenEncoder) PutNullableString(in *string) error {
 }
 
 // PutStringArray increments length for a string array.
-func (e *LenEncoder) PutStringArray(in []string) error {
+func (e *lenEncoder) PutStringArray(in []string) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
 		return err
@@ -165,7 +165,7 @@ func (e *LenEncoder) PutStringArray(in []string) error {
 }
 
 // PutInt32Array increments length for an int32 array.
-func (e *LenEncoder) PutInt32Array(in []int32) error {
+func (e *lenEncoder) PutInt32Array(in []int32) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
 		return err
@@ -175,7 +175,7 @@ func (e *LenEncoder) PutInt32Array(in []int32) error {
 }
 
 // PutInt64Array increments length for an int64 array.
-func (e *LenEncoder) PutInt64Array(in []int64) error {
+func (e *lenEncoder) PutInt64Array(in []int64) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
 		return err
@@ -184,34 +184,34 @@ func (e *LenEncoder) PutInt64Array(in []int64) error {
 	return nil
 }
 
-// Push increments length based on the PushEncoder's reserved size.
-func (e *LenEncoder) Push(pe PushEncoder) {
+// Push increments length based on the pushEncoder's reserved size.
+func (e *lenEncoder) Push(pe pushEncoder) {
 	e.Length += pe.ReserveSize()
 }
 
 // Pop is a no-op.
-func (e *LenEncoder) Pop() {}
+func (e *lenEncoder) Pop() {}
 
-// ByteEncoder is a PacketEncoder that serializes data into a byte slice.
-type ByteEncoder struct {
+// byteEncoder is a packetEncoder that serializes data into a byte slice.
+type byteEncoder struct {
 	b     []byte
 	off   int
-	stack []PushEncoder
+	stack []pushEncoder
 }
 
 // Bytes returns the underlying byte slice.
-func (e *ByteEncoder) Bytes() []byte {
+func (e *byteEncoder) Bytes() []byte {
 	return e.b
 }
 
 // NewByteEncoder creates a new ByteEncoder with the given backing
 // pre-allocated byte slice.
-func NewByteEncoder(b []byte) *ByteEncoder {
-	return &ByteEncoder{b: b}
+func newByteEncoder(b []byte) *byteEncoder {
+	return &byteEncoder{b: b}
 }
 
 // PutBool serializes a bool.
-func (e *ByteEncoder) PutBool(in bool) {
+func (e *byteEncoder) PutBool(in bool) {
 	if in {
 		e.b[e.off] = byte(int8(1))
 	}
@@ -219,44 +219,44 @@ func (e *ByteEncoder) PutBool(in bool) {
 }
 
 // PutInt8 serializes an int8.
-func (e *ByteEncoder) PutInt8(in int8) {
+func (e *byteEncoder) PutInt8(in int8) {
 	e.b[e.off] = byte(in)
 	e.off++
 }
 
 // PutInt16 serializes an int16.
-func (e *ByteEncoder) PutInt16(in int16) {
-	Encoding.PutUint16(e.b[e.off:], uint16(in))
+func (e *byteEncoder) PutInt16(in int16) {
+	encoding.PutUint16(e.b[e.off:], uint16(in))
 	e.off += 2
 }
 
 // PutInt32 serializes an int32.
-func (e *ByteEncoder) PutInt32(in int32) {
-	Encoding.PutUint32(e.b[e.off:], uint32(in))
+func (e *byteEncoder) PutInt32(in int32) {
+	encoding.PutUint32(e.b[e.off:], uint32(in))
 	e.off += 4
 }
 
 // PutInt64 serializes an int64.
-func (e *ByteEncoder) PutInt64(in int64) {
-	Encoding.PutUint64(e.b[e.off:], uint64(in))
+func (e *byteEncoder) PutInt64(in int64) {
+	encoding.PutUint64(e.b[e.off:], uint64(in))
 	e.off += 8
 }
 
 // PutArrayLength serializes an array length as an int32.
-func (e *ByteEncoder) PutArrayLength(in int) error {
+func (e *byteEncoder) PutArrayLength(in int) error {
 	e.PutInt32(int32(in))
 	return nil
 }
 
 // PutRawBytes serializes a byte slice.
-func (e *ByteEncoder) PutRawBytes(in []byte) error {
+func (e *byteEncoder) PutRawBytes(in []byte) error {
 	copy(e.b[e.off:], in)
 	e.off += len(in)
 	return nil
 }
 
 // PutBytes serializes a size-prefixed byte slice.
-func (e *ByteEncoder) PutBytes(in []byte) error {
+func (e *byteEncoder) PutBytes(in []byte) error {
 	if in == nil {
 		e.PutInt32(-1)
 		return nil
@@ -268,7 +268,7 @@ func (e *ByteEncoder) PutBytes(in []byte) error {
 }
 
 // PutString serializes a size-prefixed string.
-func (e *ByteEncoder) PutString(in string) error {
+func (e *byteEncoder) PutString(in string) error {
 	e.PutInt16(int16(len(in)))
 	copy(e.b[e.off:], in)
 	e.off += len(in)
@@ -276,7 +276,7 @@ func (e *ByteEncoder) PutString(in string) error {
 }
 
 // PutNullableString serializes a nullable string.
-func (e *ByteEncoder) PutNullableString(in *string) error {
+func (e *byteEncoder) PutNullableString(in *string) error {
 	if in == nil {
 		e.PutInt16(-1)
 		return nil
@@ -285,7 +285,7 @@ func (e *ByteEncoder) PutNullableString(in *string) error {
 }
 
 // PutStringArray serializes a string array.
-func (e *ByteEncoder) PutStringArray(in []string) error {
+func (e *byteEncoder) PutStringArray(in []string) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
 		return err
@@ -301,7 +301,7 @@ func (e *ByteEncoder) PutStringArray(in []string) error {
 }
 
 // PutInt32Array serializes an int32 array.
-func (e *ByteEncoder) PutInt32Array(in []int32) error {
+func (e *byteEncoder) PutInt32Array(in []int32) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
 		return err
@@ -313,7 +313,7 @@ func (e *ByteEncoder) PutInt32Array(in []int32) error {
 }
 
 // PutInt64Array serializes an int64 array.
-func (e *ByteEncoder) PutInt64Array(in []int64) error {
+func (e *byteEncoder) PutInt64Array(in []int64) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
 		return err
@@ -324,16 +324,16 @@ func (e *ByteEncoder) PutInt64Array(in []int64) error {
 	return nil
 }
 
-// Push adds the given PushEncoder to the stack and saves the current offset
+// Push adds the given pushEncoder to the stack and saves the current offset
 // position.
-func (e *ByteEncoder) Push(pe PushEncoder) {
+func (e *byteEncoder) Push(pe pushEncoder) {
 	pe.SaveOffset(e.off)
 	e.off += pe.ReserveSize()
 	e.stack = append(e.stack, pe)
 }
 
-// Pop the stack and run the popped PushEncoder on the serialized data.
-func (e *ByteEncoder) Pop() {
+// Pop the stack and run the popped pushEncoder on the serialized data.
+func (e *byteEncoder) Pop() {
 	// this is go's ugly pop pattern (the inverse of append)
 	pe := e.stack[len(e.stack)-1]
 	e.stack = e.stack[:len(e.stack)-1]
diff --git a/server/commitlog/interface.go b/server/commitlog/interface.go
index 76e2d3e..9058a24 100644
--- a/server/commitlog/interface.go
+++ b/server/commitlog/interface.go
@@ -1,7 +1,5 @@
 package commitlog
 
-import "github.com/liftbridge-io/liftbridge/server/proto"
-
 // CommitLog is the durable write-ahead log interface used to back each stream.
 type CommitLog interface {
 	// Delete closes the log and removes all data associated with it from the
@@ -53,7 +51,7 @@ type CommitLog interface {
 
 	// Append writes the given batch of messages to the log and returns their
 	// corresponding offsets in the log.
-	Append(msg []*proto.Message) ([]int64, error)
+	Append(msg []*Message) ([]int64, error)
 
 	// AppendMessageSet writes the given message set data to the log and
 	// returns the corresponding offsets in the log.
diff --git a/server/commitlog/message.go b/server/commitlog/message.go
index 147cd36..e74ee86 100644
--- a/server/commitlog/message.go
+++ b/server/commitlog/message.go
@@ -1,27 +1,106 @@
 package commitlog
 
-import "github.com/liftbridge-io/liftbridge/server/proto"
+import (
+	"errors"
+	"hash/crc32"
 
-// Message is a message read from the log.
-type Message []byte
+	client "github.com/liftbridge-io/liftbridge-api/go"
+)
+
+var crc32cTable = crc32.MakeTable(crc32.Castagnoli)
+
+// Message is the object that gets serialized and written to the log.
+type Message struct {
+	Crc        int32
+	MagicByte  int8
+	Attributes int8
+	Key        []byte
+	Value      []byte
+	Headers    map[string][]byte
+
+	// Transient fields
+	Timestamp     int64
+	LeaderEpoch   uint64
+	AckInbox      string
+	CorrelationID string
+	AckPolicy     client.AckPolicy
+}
+
+// Encode the Message into the packetEncoder.
+func (m *Message) Encode(e packetEncoder) error {
+	e.Push(&crcField{})
+	e.PutInt8(m.MagicByte)
+	e.PutInt8(m.Attributes)
+	if err := e.PutBytes(m.Key); err != nil {
+		return err
+	}
+	if err := e.PutBytes(m.Value); err != nil {
+		return err
+	}
+	e.PutInt16(int16(len(m.Headers)))
+	for key, header := range m.Headers {
+		if err := e.PutString(key); err != nil {
+			return err
+		}
+		if err := e.PutBytes(header); err != nil {
+			return err
+		}
+	}
+	e.Pop()
+	return nil
+}
+
+// crcField is used to perform a CRC32 check on a message.
+type crcField struct {
+	StartOffset int
+}
+
+// SaveOffset sets the position to fill the CRC digest.
+func (f *crcField) SaveOffset(in int) {
+	f.StartOffset = in
+}
+
+// ReserveSize sets the number of bytes to reserve for the CRC digest.
+func (f *crcField) ReserveSize() int {
+	return 4
+}
+
+// Fill sets the CRC digest.
+func (f *crcField) Fill(curOffset int, buf []byte) error {
+	crc := crc32.Checksum(buf[f.StartOffset+4:curOffset], crc32cTable)
+	encoding.PutUint32(buf[f.StartOffset:], crc)
+	return nil
+}
+
+// Check the CRC digest.
+func (f *crcField) Check(curOffset int, buf []byte) error {
+	crc := crc32.Checksum(buf[f.StartOffset+4:curOffset], crc32cTable)
+	if crc != encoding.Uint32(buf[f.StartOffset:]) {
+		return errors.New("crc didn't match")
+	}
+	return nil
+}
+
+// SerializedMessage is a serialized message read from the log.
+type SerializedMessage []byte
 
 // Crc returns the CRC32 digest of the message.
-func (m Message) Crc() uint32 {
-	return proto.Encoding.Uint32(m)
+func (m SerializedMessage) Crc() uint32 {
+	return encoding.Uint32(m)
 }
 
 // MagicByte returns the byte used for encoding protocol version detection.
-func (m Message) MagicByte() int8 {
+func (m SerializedMessage) MagicByte() int8 {
 	return int8(m[4])
 }
 
 // Attributes returns the byte used for message flags.
-func (m Message) Attributes() int8 {
+func (m SerializedMessage) Attributes() int8 {
 	return int8(m[5])
 }
 
 // Key returns the message key.
-func (m Message) Key() []byte {
+func (m SerializedMessage) Key() []byte {
 	start, end, size := m.keyOffsets()
 	if size == -1 {
 		return nil
@@ -30,7 +109,7 @@ func (m Message) Key() []byte {
 }
 
 // Value returns the message value.
-func (m Message) Value() []byte {
+func (m SerializedMessage) Value() []byte {
 	start, end, size := m.valueOffsets()
 	if size == -1 {
 		return nil
@@ -39,20 +118,20 @@ func (m Message) Value() []byte {
 }
 
 // Headers returns the message headers map.
-func (m Message) Headers() map[string][]byte {
+func (m SerializedMessage) Headers() map[string][]byte {
 	var (
 		_, valueEnd, _ = m.valueOffsets()
 		n              = valueEnd
-		numHeaders     = proto.Encoding.Uint16(m[n:])
+		numHeaders     = encoding.Uint16(m[n:])
 		headers        = make(map[string][]byte, numHeaders)
 	)
 	n += 2
 	for i := uint16(0); i < numHeaders; i++ {
-		keySize := proto.Encoding.Uint16(m[n:])
+		keySize := encoding.Uint16(m[n:])
 		n += 2
 		key := string(m[n : n+int32(keySize)])
 		n += int32(keySize)
-		valueSize := proto.Encoding.Uint32(m[n:])
+		valueSize := encoding.Uint32(m[n:])
 		n += 4
 		value := m[n : n+int32(valueSize)]
 		n += int32(valueSize)
@@ -61,9 +140,9 @@ func (m Message) Headers() map[string][]byte {
 	return headers
 }
 
-func (m Message) keyOffsets() (start, end, size int32) {
+func (m SerializedMessage) keyOffsets() (start, end, size int32) {
 	start = 6
-	size = int32(proto.Encoding.Uint32(m[start:]))
+	size = int32(encoding.Uint32(m[start:]))
 	end = start + 4
 	if size != -1 {
 		end += size
@@ -71,10 +150,10 @@ func (m Message) keyOffsets() (start, end, size int32) {
 	return
 }
 
-func (m Message) valueOffsets() (start, end, size int32) {
+func (m SerializedMessage) valueOffsets() (start, end, size int32) {
 	_, keyEnd, _ := m.keyOffsets()
 	start = keyEnd
-	size = int32(proto.Encoding.Uint32(m[start:]))
+	size = int32(encoding.Uint32(m[start:]))
 	end = start + 4
 	if size != -1 {
 		end += size
diff --git a/server/commitlog/message_set.go b/server/commitlog/message_set.go
index be7bdb2..5ac9359 100644
--- a/server/commitlog/message_set.go
+++ b/server/commitlog/message_set.go
@@ -8,8 +8,6 @@ import (
 	"hash/crc32"
 
 	"github.com/pkg/errors"
-
-	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 const (
@@ -50,7 +48,7 @@ func entriesForMessageSet(basePos int64, ms []byte) []*entry {
 	return entries
 }
 
-func newMessageSetFromProto(baseOffset, basePos int64, msgs []*proto.Message) (
+func newMessageSetFromProto(baseOffset, basePos int64, msgs []*Message) (
 	messageSet, []*entry, error) {
 
 	var (
@@ -59,7 +57,7 @@ func newMessageSetFromProto(baseOffset, basePos int64, msgs []*proto.Message) (
 		n       int32
 	)
 	for i, m := range msgs {
-		data, err := proto.Encode(m)
+		data, err := encode(m)
 		if err != nil {
 			panic(err)
 		}
@@ -68,19 +66,19 @@ func newMessageSetFromProto(baseOffset, basePos int64, msgs []*proto.Message) (
 			relPos = int64(n)
 			offset = int64(i) + baseOffset
 		)
-		if err := binary.Write(buf, proto.Encoding, uint64(offset)); err != nil {
+		if err := binary.Write(buf, encoding, uint64(offset)); err != nil {
 			return nil, nil, err
 		}
 		n += 8
-		if err := binary.Write(buf, proto.Encoding, uint64(m.Timestamp)); err != nil {
+		if err := binary.Write(buf, encoding, uint64(m.Timestamp)); err != nil {
 			return nil, nil, err
 		}
 		n += 8
-		if err := binary.Write(buf, proto.Encoding, m.LeaderEpoch); err != nil {
+		if err := binary.Write(buf, encoding, m.LeaderEpoch); err != nil {
 			return nil, nil, err
 		}
 		n += 8
-		if err := binary.Write(buf, proto.Encoding, uint32(len)); err != nil {
+		if err := binary.Write(buf, encoding, uint32(len)); err != nil {
 			return nil, nil, err
 		}
 		n += 4
@@ -103,24 +101,24 @@ func newMessageSetFromProto(baseOffset, basePos int64, msgs []*proto.Message) (
 // available. It returns the Message in addition to its offset, timestamp, and
 // leader epoch. This may return uncommitted messages if the reader was created
 // with the uncommitted flag set to true.
-func readMessage(ctx context.Context, reader contextReader, headersBuf []byte) (Message, int64, int64, uint64, error) {
+func readMessage(ctx context.Context, reader contextReader, headersBuf []byte) (SerializedMessage, int64, int64, uint64, error) {
 	if _, err := reader.Read(ctx, headersBuf); err != nil {
 		return nil, 0, 0, 0, errors.Wrap(err, "failed to read message headers")
 	}
 	var (
-		offset      = int64(proto.Encoding.Uint64(headersBuf[offsetPos:]))
-		timestamp   = int64(proto.Encoding.Uint64(headersBuf[timestampPos:]))
-		leaderEpoch = proto.Encoding.Uint64(headersBuf[leaderEpochPos:])
-		size        = proto.Encoding.Uint32(headersBuf[sizePos:])
+		offset      = int64(encoding.Uint64(headersBuf[offsetPos:]))
+		timestamp   = int64(encoding.Uint64(headersBuf[timestampPos:]))
+		leaderEpoch = encoding.Uint64(headersBuf[leaderEpochPos:])
+		size        = encoding.Uint32(headersBuf[sizePos:])
 		buf         = make([]byte, int(size))
 	)
 	if _, err := reader.Read(ctx, buf); err != nil {
 		return nil, 0, 0, 0, errors.Wrap(err, "failed to ready message payload")
 	}
-	m := Message(buf)
+	m := SerializedMessage(buf)
 	// Check the CRC on the message.
 	crc := m.Crc()
-	if c := crc32.ChecksumIEEE(m[4:]); crc != c {
+	if c := crc32.Checksum(m[4:], crc32cTable); crc != c {
 		// If the CRC doesn't match, data on disk is corrupted which means the
 		// server is in an unrecoverable state.
 		panic(fmt.Errorf("Read corrupted data, expected CRC: 0x%08x, got: 0x%08x", crc, c))
@@ -129,25 +127,25 @@ func readMessage(ctx context.Context, reader contextReader, headersBuf []byte) (
 }
 
 func (ms messageSet) Offset() int64 {
-	return int64(proto.Encoding.Uint64(ms[offsetPos : offsetPos+8]))
+	return int64(encoding.Uint64(ms[offsetPos : offsetPos+8]))
 }
 
 func (ms messageSet) Timestamp() int64 {
-	return int64(proto.Encoding.Uint64(ms[timestampPos : timestampPos+8]))
+	return int64(encoding.Uint64(ms[timestampPos : timestampPos+8]))
 }
 
 func (ms messageSet) LeaderEpoch() uint64 {
-	return proto.Encoding.Uint64(ms[leaderEpochPos : leaderEpochPos+8])
+	return encoding.Uint64(ms[leaderEpochPos : leaderEpochPos+8])
 }
 
 func (ms messageSet) Size() int32 {
-	return int32(proto.Encoding.Uint32(ms[sizePos : sizePos+4]))
+	return int32(encoding.Uint32(ms[sizePos : sizePos+4]))
 }
 
-func (ms messageSet) Message() Message {
+func (ms messageSet) Message() SerializedMessage {
 	if len(ms) <= msgSetHeaderLen {
 		return nil
 	}
 	size := ms.Size()
-	return Message(ms[msgSetHeaderLen : msgSetHeaderLen+size])
+	return SerializedMessage(ms[msgSetHeaderLen : msgSetHeaderLen+size])
 }
diff --git a/server/commitlog/reader.go b/server/commitlog/reader.go
index df17467..a0fcdb0 100644
--- a/server/commitlog/reader.go
+++ b/server/commitlog/reader.go
@@ -44,16 +44,16 @@ func (l *commitLog) NewReader(offset int64, uncommitted bool) (*Reader, error) {
 }
 
 // ReadMessage reads a single message from the underlying CommitLog or blocks
-// until one is available. It returns the Message in addition to its offset,
-// timestamp, and leader epoch. This may return uncommitted messages if the
-// reader was created with the uncommitted flag set to true.
+// until one is available. It returns the SerializedMessage in addition to its
+// offset, timestamp, and leader epoch. This may return uncommitted messages if
+// the reader was created with the uncommitted flag set to true.
 //
 // ReadMessage should not be called concurrently, and the headersBuf slice
 // should have a capacity of at least 28.
 //
 // TODO: Should this just return a MessageSet directly instead of a Message and
 // the MessageSet header values?
-func (r *Reader) ReadMessage(ctx context.Context, headersBuf []byte) (Message, int64, int64, uint64, error) {
+func (r *Reader) ReadMessage(ctx context.Context, headersBuf []byte) (SerializedMessage, int64, int64, uint64, error) {
 RETRY:
 	msg, offset, timestamp, leaderEpoch, err := readMessage(ctx, r.ctxReader, headersBuf)
 	if err != nil {
diff --git a/server/partition.go b/server/partition.go
index a5e8e98..f61b7ab 100644
--- a/server/partition.go
+++ b/server/partition.go
@@ -1,7 +1,6 @@
 package server
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"math/rand"
@@ -23,16 +22,9 @@ import (
 // message processing loop.
 const recvChannelSize = 64 * 1024
 
-var (
-	// envelopeCookie is a magic value that indicates if a NATS message is a
-	// structured message protobuf.
-	envelopeCookie    = []byte("LIFT")
-	envelopeCookieLen = len(envelopeCookie)
-
-	// timestamp returns the current time in Unix nanoseconds. This function
-	// exists for mocking purposes.
-	timestamp = func() int64 { return time.Now().UnixNano() }
-)
+// timestamp returns the current time in Unix nanoseconds. This function exists
+// for mocking purposes.
+var timestamp = func() int64 { return time.Now().UnixNano() }
 
 // replica tracks the latest log offset for a particular partition replica.
 type replica struct {
@@ -549,7 +541,7 @@ func (p *partition) messageProcessingLoop(recvChan <-chan *nats.Msg, stop <-chan
 		msg       *nats.Msg
 		batchSize = p.srv.config.BatchMaxMessages
 		batchWait = p.srv.config.BatchWaitTime
-		msgBatch  = make([]*proto.Message, 0, batchSize)
+		msgBatch  = make([]*commitlog.Message, 0, batchSize)
 	)
 	for {
 		msgBatch = msgBatch[:0]
@@ -613,7 +605,7 @@ func (p *partition) messageProcessingLoop(recvChan <-chan *nats.Msg, stop <-chan
 // processPendingMessage sends an ack if the message's AckPolicy is LEADER and
 // adds the pending message to the commit queue. Messages are removed from the
 // queue and committed when the entire ISR has replicated them.
-func (p *partition) processPendingMessage(offset int64, msg *proto.Message) {
+func (p *partition) processPendingMessage(offset int64, msg *commitlog.Message) {
 	ack := &client.Ack{
 		Stream:           p.Stream,
 		PartitionSubject: p.Subject,
@@ -736,7 +728,7 @@ func (p *partition) sendAck(ack *client.Ack) {
 	if ack.AckInbox == "" {
 		return
 	}
-	data, err := ack.Marshal()
+	data, err := proto.MarshalEnvelope(ack)
 	if err != nil {
 		panic(err)
 	}
@@ -1099,26 +1091,20 @@ func (p *partition) getSubject() string {
 }
 
 // getMessage converts the given payload into a client Message if it is one.
-// This is indicated by the presence of the envelope cookie. If it is not, nil
-// is returned.
+// This is indicated by the presence of the envelope magic number. If it is
+// not, nil is returned.
 func getMessage(data []byte) *client.Message {
-	if len(data) <= 4 {
-		return nil
-	}
-	if !bytes.Equal(data[0:4], envelopeCookie) {
-		return nil
-	}
-	msg := &client.Message{}
-	if err := msg.Unmarshal(data[4:]); err != nil {
+	msg := new(client.Message)
+	if err := proto.UnmarshalEnvelope(data, msg); err != nil {
 		return nil
 	}
 	return msg
 }
 
-// natsToProtoMessage converts the given NATS message to a proto Message.
-func natsToProtoMessage(msg *nats.Msg, leaderEpoch uint64) *proto.Message {
+// natsToProtoMessage converts the given NATS message to a commit log Message.
+func natsToProtoMessage(msg *nats.Msg, leaderEpoch uint64) *commitlog.Message {
 	message := getMessage(msg.Data)
-	m := &proto.Message{
+	m := &commitlog.Message{
 		MagicByte:   1,
 		Timestamp:   timestamp(),
 		LeaderEpoch: leaderEpoch,
diff --git a/server/proto/crc_field.go b/server/proto/crc_field.go
deleted file mode 100644
index 22282a6..0000000
--- a/server/proto/crc_field.go
+++ /dev/null
@@ -1,37 +0,0 @@
-package proto
-
-import (
-	"errors"
-	"hash/crc32"
-)
-
-// CRCField is used to perform a CRC32 check on a message.
-type CRCField struct {
-	StartOffset int
-}
-
-// SaveOffset sets the position to fill the CRC digest.
-func (f *CRCField) SaveOffset(in int) {
-	f.StartOffset = in
-}
-
-// ReserveSize sets the number of bytes to reserve for the CRC digest.
-func (f *CRCField) ReserveSize() int {
-	return 4
-}
-
-// Fill sets the CRC digest.
-func (f *CRCField) Fill(curOffset int, buf []byte) error {
-	crc := crc32.ChecksumIEEE(buf[f.StartOffset+4 : curOffset])
-	Encoding.PutUint32(buf[f.StartOffset:], crc)
-	return nil
-}
-
-// Check the CRC digest.
-func (f *CRCField) Check(curOffset int, buf []byte) error {
-	crc := crc32.ChecksumIEEE(buf[f.StartOffset+4 : curOffset])
-	if crc != Encoding.Uint32(buf[f.StartOffset:]) {
-		return errors.New("crc didn't match")
-	}
-	return nil
-}
diff --git a/server/proto/envelope.go b/server/proto/envelope.go
new file mode 100644
index 0000000..f1fb063
--- /dev/null
+++ b/server/proto/envelope.go
@@ -0,0 +1,105 @@
+package proto
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"hash/crc32"
+
+	pb "github.com/golang/protobuf/proto"
+)
+
+const (
+	// envelopeProtoV0 is version 0 of the envelope protocol.
+	envelopeProtoV0 = 0x00
+
+	// envelopeMinHeaderLen is the minimum length of the envelope header, i.e.
+	// without CRC-32C set.
+	envelopeMinHeaderLen = 8
+)
+
+var (
+	// Encoding is the byte order to use for protocol serialization.
+	Encoding = binary.BigEndian
+
+	// envelopeMagicNumber is a value that indicates if a NATS message is a
+	// structured message protobuf. This was chosen by random but deliberately
+	// restricted to invalid UTF-8 to reduce the chance of a collision. This
+	// was also verified to not match known file signatures.
+	envelopeMagicNumber    = []byte{0xB9, 0x0E, 0x43, 0xB4}
+	envelopeMagicNumberLen = len(envelopeMagicNumber)
+
+	crc32cTable = crc32.MakeTable(crc32.Castagnoli)
+)
+
+// MarshalEnvelope serializes a protobuf message into the Liftbridge envelope
+// wire format.
+func MarshalEnvelope(msg pb.Message) ([]byte, error) {
+	data, err := pb.Marshal(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	var (
+		buf       = make([]byte, envelopeMagicNumberLen+4+len(data))
+		pos       = 0
+		headerLen = envelopeMinHeaderLen
+	)
+	copy(buf[pos:], envelopeMagicNumber)
+	pos += envelopeMagicNumberLen
+	buf[pos] = envelopeProtoV0 // Version
+	pos++
+	buf[pos] = byte(headerLen) // HeaderLen
+	pos++
+	buf[pos] = 0x00 // Flags
+	pos++
+	buf[pos] = 0x00 // Reserved
+	pos++
+	if pos != headerLen {
+		panic(fmt.Sprintf("Payload position (%d) does not match expected HeaderLen (%d)",
+			pos, headerLen))
+	}
+	copy(buf[pos:], data)
+	return buf, nil
+}
+
+// UnmarshalEnvelope deserializes a Liftbridge envelope into a protobuf
+// message.
+func UnmarshalEnvelope(data []byte, msg pb.Message) error {
+	if len(data) < envelopeMinHeaderLen {
+		return errors.New("data missing envelope header")
+	}
+	if !bytes.Equal(data[:envelopeMagicNumberLen], envelopeMagicNumber) {
+		return errors.New("unexpected envelope magic number")
+	}
+	if data[4] != envelopeProtoV0 {
+		return fmt.Errorf("unknown envelope protocol: %v", data[4])
+	}
+
+	var (
+		headerLen = int(data[5])
+		flags     = data[6]
+		payload   = data[headerLen:]
+	)
+
+	// Check CRC.
+	if hasBit(flags, 0) {
+		// Make sure there is a CRC present.
+		if headerLen != envelopeMinHeaderLen+4 {
+			return errors.New("incorrect envelope header size")
+		}
+		crc := Encoding.Uint32(data[envelopeMinHeaderLen:headerLen])
+		if c := crc32.Checksum(payload, crc32cTable); c != crc {
+			return fmt.Errorf("crc mismatch: expected %d, got %d", crc, c)
+		}
+	}
+
+	return pb.Unmarshal(payload, msg)
+}
+
+// hasBit checks if the given bit position is set on the provided byte.
+func hasBit(n byte, pos uint8) bool {
+	val := n & (1 << pos)
+	return (val > 0)
+}
diff --git a/server/proto/message.go b/server/proto/message.go
deleted file mode 100644
index af81350..0000000
--- a/server/proto/message.go
+++ /dev/null
@@ -1,46 +0,0 @@
-package proto
-
-import (
-	client "github.com/liftbridge-io/liftbridge-api/go"
-)
-
-// Message is the serialized object that gets written to the log.
-type Message struct {
-	Crc        int32
-	MagicByte  int8
-	Attributes int8
-	Key        []byte
-	Value      []byte
-	Headers    map[string][]byte
-
-	// Transient fields
-	Timestamp     int64
-	LeaderEpoch   uint64
-	AckInbox      string
-	CorrelationID string
-	AckPolicy     client.AckPolicy
-}
-
-// Encode the Message into the PacketEncoder.
-func (m *Message) Encode(e PacketEncoder) error {
-	e.Push(&CRCField{})
-	e.PutInt8(m.MagicByte)
-	e.PutInt8(m.Attributes)
-	if err := e.PutBytes(m.Key); err != nil {
-		return err
-	}
-	if err := e.PutBytes(m.Value); err != nil {
-		return err
-	}
-	e.PutInt16(int16(len(m.Headers)))
-	for key, header := range m.Headers {
-		if err := e.PutString(key); err != nil {
-			return err
-		}
-		if err := e.PutBytes(header); err != nil {
-			return err
-		}
-	}
-	e.Pop()
-	return nil
-}
diff --git a/server/replicator.go b/server/replicator.go
index 0caf135..6a1ad49 100644
--- a/server/replicator.go
+++ b/server/replicator.go
@@ -226,7 +226,7 @@ func (r *replicator) replicate(
 
 	var (
 		newestOffset = r.partition.log.NewestOffset()
-		message      commitlog.Message
+		message      commitlog.SerializedMessage
 		err          error
 	)
 	for offset < newestOffset && r.writer.Len() < replicationMaxSize {
diff --git a/website/i18n/en.json b/website/i18n/en.json
index 01e0c33..3a2824d 100644
--- a/website/i18n/en.json
+++ b/website/i18n/en.json
@@ -20,6 +20,9 @@
       "deployment": {
         "title": "Deployment"
       },
+      "envelope-protocol": {
+        "title": "Envelope Protocol"
+      },
       "faq": {
         "title": "Frequently Asked Questions"
       },
diff --git a/website/sidebars.json b/website/sidebars.json
index 90a734c..a323923 100755
--- a/website/sidebars.json
+++ b/website/sidebars.json
@@ -14,7 +14,8 @@
         "deployment"
     ],
     "Technical Deep Dive": [
-        "replication-protocol"
+        "replication-protocol",
+        "envelope-protocol"
     ],
     "Client Libraries": [
         "clients",


Test patch:
diff --git a/server/api_test.go b/server/api_test.go
index 99726d8..0af11d2 100644
--- a/server/api_test.go
+++ b/server/api_test.go
@@ -23,10 +23,10 @@ type message struct {
 	Offset int64
 }
 
-func assertMsg(t *testing.T, expected *message, msg *proto.Message) {
-	require.Equal(t, expected.Offset, msg.Offset)
-	require.Equal(t, expected.Key, msg.Key)
-	require.Equal(t, expected.Value, msg.Value)
+func assertMsg(t *testing.T, expected *message, msg lift.Message) {
+	require.Equal(t, expected.Offset, msg.Offset())
+	require.Equal(t, expected.Key, msg.Key())
+	require.Equal(t, expected.Value, msg.Value())
 }
 
 // Ensure creating a stream works, and it returns an error when creating the
@@ -302,7 +302,7 @@ func TestStreamPublishSubscribe(t *testing.T) {
 	i := 0
 	ch1 := make(chan struct{})
 	ch2 := make(chan struct{})
-	err = client.Subscribe(context.Background(), name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(context.Background(), name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
 		expect := expected[i]
 		assertMsg(t, expect, msg)
@@ -318,7 +318,7 @@ func TestStreamPublishSubscribe(t *testing.T) {
 
 	// Publish messages.
 	for i := 0; i < num; i++ {
-		_, err = client.Publish(context.Background(), subject, expected[i].Value,
+		_, err = client.Publish(context.Background(), name, expected[i].Value,
 			lift.Key(expected[i].Key))
 		require.NoError(t, err)
 	}
@@ -339,7 +339,7 @@ func TestStreamPublishSubscribe(t *testing.T) {
 		})
 	}
 	for i := 0; i < 5; i++ {
-		_, err = client.Publish(context.Background(), subject, expected[i+num].Value,
+		_, err = client.Publish(context.Background(), name, expected[i+num].Value,
 			lift.Key(expected[i+num].Key))
 		require.NoError(t, err)
 	}
@@ -358,7 +358,7 @@ func TestStreamPublishSubscribe(t *testing.T) {
 	i = num
 	ch1 = make(chan struct{})
 	err = client2.Subscribe(context.Background(), name,
-		func(msg *proto.Message, err error) {
+		func(msg lift.Message, err error) {
 			require.NoError(t, err)
 			expect := expected[i]
 			assertMsg(t, expect, msg)
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index d0e2d26..e771316 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -9,15 +9,13 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
-
-	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 var (
 	headers = map[string][]byte{
 		"foo": []byte("bar"),
 	}
-	msgs = []*proto.Message{
+	msgs = []*Message{
 		{Value: []byte("one"), Timestamp: 1, LeaderEpoch: 42, Headers: headers},
 		{Value: []byte("two"), Timestamp: 2, LeaderEpoch: 42, Headers: headers},
 		{Value: []byte("three"), Timestamp: 3, LeaderEpoch: 42, Headers: headers},
@@ -97,12 +95,12 @@ func TestCommitLogRecover(t *testing.T) {
 
 			// Append some messages.
 			numMsgs := 10
-			msgs := make([]*proto.Message, numMsgs)
+			msgs := make([]*Message, numMsgs)
 			for i := 0; i < numMsgs; i++ {
-				msgs[i] = &proto.Message{Value: []byte(strconv.Itoa(i))}
+				msgs[i] = &Message{Value: []byte(strconv.Itoa(i))}
 			}
 			for _, msg := range msgs {
-				_, err := l.Append([]*proto.Message{msg})
+				_, err := l.Append([]*Message{msg})
 				require.NoError(t, err)
 			}
 
@@ -196,9 +194,9 @@ func TestOffsets(t *testing.T) {
 	require.Equal(t, int64(-1), l.NewestOffset())
 
 	numMsgs := 5
-	msgs := make([]*proto.Message, numMsgs)
+	msgs := make([]*Message, numMsgs)
 	for i := 0; i < numMsgs; i++ {
-		msgs[i] = &proto.Message{Value: []byte(strconv.Itoa(i))}
+		msgs[i] = &Message{Value: []byte(strconv.Itoa(i))}
 	}
 	_, err := l.Append(msgs)
 	require.NoError(t, err)
@@ -254,7 +252,7 @@ func TestCleanerDeleteLeaderEpochOffsets(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -263,7 +261,7 @@ func TestCleanerDeleteLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i + 5)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 2,
@@ -272,7 +270,7 @@ func TestCleanerDeleteLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i + 10)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 3,
@@ -319,7 +317,7 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Key:         []byte("foo"),
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
@@ -329,7 +327,7 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Key:         []byte("bar"),
 			Value:       []byte(strconv.Itoa(i + 5)),
 			Timestamp:   time.Now().UnixNano(),
@@ -339,7 +337,7 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Key:         []byte("baz"),
 			Value:       []byte(strconv.Itoa(i + 10)),
 			Timestamp:   time.Now().UnixNano(),
@@ -386,12 +384,12 @@ func TestOffsetForTimestamp(t *testing.T) {
 
 	// Append some messages.
 	numMsgs := 10
-	msgs := make([]*proto.Message, numMsgs)
+	msgs := make([]*Message, numMsgs)
 	for i := 0; i < numMsgs; i++ {
-		msgs[i] = &proto.Message{Value: []byte(strconv.Itoa(i)), Timestamp: int64(i * 10)}
+		msgs[i] = &Message{Value: []byte(strconv.Itoa(i)), Timestamp: int64(i * 10)}
 	}
 	for _, msg := range msgs {
-		_, err := l.Append([]*proto.Message{msg})
+		_, err := l.Append([]*Message{msg})
 		require.NoError(t, err)
 	}
 
@@ -439,7 +437,7 @@ func TestTruncate(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -448,7 +446,7 @@ func TestTruncate(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i + 5)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 2,
@@ -457,7 +455,7 @@ func TestTruncate(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i + 10)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 3,
@@ -491,7 +489,7 @@ func TestNotifyLEOMismatch(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -501,7 +499,7 @@ func TestNotifyLEOMismatch(t *testing.T) {
 
 	// Get current log end offset and then add another message.
 	leo := l.NewestOffset()
-	_, err := l.Append([]*proto.Message{{
+	_, err := l.Append([]*Message{{
 		Value:       []byte(strconv.Itoa(5)),
 		Timestamp:   time.Now().UnixNano(),
 		LeaderEpoch: 1,
@@ -531,7 +529,7 @@ func TestNotifyLEONewData(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{{
+		_, err := l.Append([]*Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -553,7 +551,7 @@ func TestNotifyLEONewData(t *testing.T) {
 	}
 
 	// Add another message.
-	_, err := l.Append([]*proto.Message{{
+	_, err := l.Append([]*Message{{
 		Value:       []byte(strconv.Itoa(5)),
 		Timestamp:   time.Now().UnixNano(),
 		LeaderEpoch: 1,
diff --git a/server/commitlog/compact_cleaner_test.go b/server/commitlog/compact_cleaner_test.go
index b409c88..960486e 100644
--- a/server/commitlog/compact_cleaner_test.go
+++ b/server/commitlog/compact_cleaner_test.go
@@ -7,8 +7,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-
-	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 type keyValue struct {
@@ -18,7 +16,7 @@ type keyValue struct {
 
 type expectedMsg struct {
 	Offset int64
-	Msg    *proto.Message
+	Msg    *Message
 }
 
 // Ensure Compact is a no-op when there are no segments.
@@ -75,11 +73,11 @@ func TestCompactCleaner(t *testing.T) {
 	require.NoError(t, l.Clean())
 
 	expected := []*expectedMsg{
-		{Offset: 4, Msg: &proto.Message{Key: []byte("bar"), Value: []byte("second")}},
-		{Offset: 7, Msg: &proto.Message{Key: []byte("qux"), Value: []byte("first")}},
-		{Offset: 8, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("fourth")}},
+		{Offset: 4, Msg: &Message{Key: []byte("bar"), Value: []byte("second")}},
+		{Offset: 7, Msg: &Message{Key: []byte("qux"), Value: []byte("first")}},
+		{Offset: 8, Msg: &Message{Key: []byte("foo"), Value: []byte("fourth")}},
 		// This one is present because it's in the active segment.
-		{Offset: 9, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("third")}},
+		{Offset: 9, Msg: &Message{Key: []byte("baz"), Value: []byte("third")}},
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -125,14 +123,14 @@ func TestCompactCleanerHW(t *testing.T) {
 	require.NoError(t, l.Clean())
 
 	expected := []*expectedMsg{
-		{Offset: 3, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("third")}},
-		{Offset: 4, Msg: &proto.Message{Key: []byte("bar"), Value: []byte("second")}},
-		{Offset: 5, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("first")}},
+		{Offset: 3, Msg: &Message{Key: []byte("foo"), Value: []byte("third")}},
+		{Offset: 4, Msg: &Message{Key: []byte("bar"), Value: []byte("second")}},
+		{Offset: 5, Msg: &Message{Key: []byte("baz"), Value: []byte("first")}},
 		// These are retained because they are after the HW.
-		{Offset: 6, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("second")}},
-		{Offset: 7, Msg: &proto.Message{Key: []byte("qux"), Value: []byte("first")}},
-		{Offset: 8, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("fourth")}},
-		{Offset: 9, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("third")}},
+		{Offset: 6, Msg: &Message{Key: []byte("baz"), Value: []byte("second")}},
+		{Offset: 7, Msg: &Message{Key: []byte("qux"), Value: []byte("first")}},
+		{Offset: 8, Msg: &Message{Key: []byte("foo"), Value: []byte("fourth")}},
+		{Offset: 9, Msg: &Message{Key: []byte("baz"), Value: []byte("third")}},
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -171,10 +169,10 @@ func TestCompactCleanerNoKeys(t *testing.T) {
 	require.NoError(t, l.Clean())
 
 	expected := []*expectedMsg{
-		{Offset: 0, Msg: &proto.Message{Value: []byte("first")}},
-		{Offset: 1, Msg: &proto.Message{Value: []byte("second")}},
-		{Offset: 2, Msg: &proto.Message{Value: []byte("third")}},
-		{Offset: 3, Msg: &proto.Message{Value: []byte("fourth")}},
+		{Offset: 0, Msg: &Message{Value: []byte("first")}},
+		{Offset: 1, Msg: &Message{Value: []byte("second")}},
+		{Offset: 2, Msg: &Message{Value: []byte("third")}},
+		{Offset: 3, Msg: &Message{Value: []byte("fourth")}},
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -282,11 +280,11 @@ func benchmarkClean(b *testing.B, segmentSize int64) {
 
 				buf := make([]byte, msgSize)
 				for i := 0; i < 200000; i++ {
-					msg := &proto.Message{
+					msg := &Message{
 						Key:   []byte(keys[rand.Intn(len(keys))]),
 						Value: buf,
 					}
-					offsets, err := l.Append([]*proto.Message{msg})
+					offsets, err := l.Append([]*Message{msg})
 					require.NoError(b, err)
 					l.SetHighWatermark(offsets[len(offsets)-1])
 				}
@@ -301,11 +299,11 @@ func benchmarkClean(b *testing.B, segmentSize int64) {
 
 func appendToLog(t *testing.T, l *commitLog, entries []keyValue, commit bool) {
 	for _, entry := range entries {
-		msg := &proto.Message{
+		msg := &Message{
 			Key:   entry.key,
 			Value: entry.value,
 		}
-		offsets, err := l.Append([]*proto.Message{msg})
+		offsets, err := l.Append([]*Message{msg})
 		require.NoError(t, err)
 		if commit {
 			l.SetHighWatermark(offsets[len(offsets)-1])
diff --git a/server/commitlog/delete_cleaner_test.go b/server/commitlog/delete_cleaner_test.go
index a61ca36..aaaa717 100644
--- a/server/commitlog/delete_cleaner_test.go
+++ b/server/commitlog/delete_cleaner_test.go
@@ -8,7 +8,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/liftbridge-io/liftbridge/server/logger"
-	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 func noopLogger() logger.Logger {
@@ -179,7 +178,7 @@ func TestDeleteCleanerAge(t *testing.T) {
 	for i := 0; i < 20; i++ {
 		segs[i] = createSegment(t, dir, int64(i), 20)
 		ms, entries, err := newMessageSetFromProto(int64(i), 0,
-			[]*proto.Message{{Timestamp: int64(i * 10)}})
+			[]*Message{{Timestamp: int64(i * 10)}})
 		require.NoError(t, err)
 		require.NoError(t, segs[i].WriteMessageSet(ms, entries))
 	}
@@ -212,7 +211,7 @@ func TestDeleteCleanerMessagesBelowAgeLimit(t *testing.T) {
 	for i := 0; i < 5; i++ {
 		expected[i] = createSegment(t, dir, int64(i), 20)
 		ms, entries, err := newMessageSetFromProto(int64(i), 0,
-			[]*proto.Message{{Timestamp: int64(i * 10)}})
+			[]*Message{{Timestamp: int64(i * 10)}})
 		require.NoError(t, err)
 		require.NoError(t, expected[i].WriteMessageSet(ms, entries))
 	}
@@ -277,7 +276,7 @@ func TestDeleteCleanerMessagesCompacted(t *testing.T) {
 
 func writeToSegment(t *testing.T, seg *segment, offset int64, data []byte) {
 	ms, entries, err := newMessageSetFromProto(int64(offset), seg.Position(),
-		[]*proto.Message{
+		[]*Message{
 			{
 				Timestamp:   time.Now().UnixNano(),
 				LeaderEpoch: 42,
diff --git a/server/commitlog/reader_test.go b/server/commitlog/reader_test.go
index 28dadc3..b68e3e1 100644
--- a/server/commitlog/reader_test.go
+++ b/server/commitlog/reader_test.go
@@ -9,8 +9,6 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
-
-	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 var segmentSizeTests = []struct {
@@ -35,9 +33,9 @@ func TestReaderUncommittedStartOffset(t *testing.T) {
 			defer cleanup()
 
 			numMsgs := 10
-			msgs := make([]*proto.Message, numMsgs)
+			msgs := make([]*Message, numMsgs)
 			for i := 0; i < numMsgs; i++ {
-				msgs[i] = &proto.Message{
+				msgs[i] = &Message{
 					Value:       []byte(strconv.Itoa(i)),
 					Timestamp:   int64(i),
 					LeaderEpoch: 42,
@@ -70,8 +68,8 @@ func TestReaderUncommittedBlockCancel(t *testing.T) {
 	defer l.Close()
 	defer cleanup()
 
-	msg := &proto.Message{Value: []byte("hi")}
-	_, err := l.Append([]*proto.Message{msg})
+	msg := &Message{Value: []byte("hi")}
+	_, err := l.Append([]*Message{msg})
 	require.NoError(t, err)
 
 	r, err := l.NewReader(0, true)
@@ -95,12 +93,12 @@ func TestReaderUncommittedBlockForSegmentWrite(t *testing.T) {
 	defer l.Close()
 	defer cleanup()
 
-	msg := &proto.Message{
+	msg := &Message{
 		Value:       []byte("hi"),
 		Timestamp:   1,
 		LeaderEpoch: 42,
 	}
-	_, err := l.Append([]*proto.Message{msg})
+	_, err := l.Append([]*Message{msg})
 	require.NoError(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -115,7 +113,7 @@ func TestReaderUncommittedBlockForSegmentWrite(t *testing.T) {
 	require.Equal(t, uint64(42), leaderEpoch)
 	compareMessages(t, msg, m)
 
-	msg = &proto.Message{
+	msg = &Message{
 		Value:       []byte("hello"),
 		Timestamp:   2,
 		LeaderEpoch: 43,
@@ -125,7 +123,7 @@ func TestReaderUncommittedBlockForSegmentWrite(t *testing.T) {
 
 	go func() {
 		time.Sleep(5 * time.Millisecond)
-		_, err := l.Append([]*proto.Message{msg})
+		_, err := l.Append([]*Message{msg})
 		require.NoError(t, err)
 		close(done)
 	}()
@@ -147,8 +145,8 @@ func TestReaderUncommittedReadError(t *testing.T) {
 	})
 	defer cleanup()
 
-	msg := &proto.Message{Value: []byte("hi")}
-	_, err := l.Append([]*proto.Message{msg})
+	msg := &Message{Value: []byte("hi")}
+	_, err := l.Append([]*Message{msg})
 	require.NoError(t, err)
 
 	r, err := l.NewReader(0, true)
@@ -173,9 +171,9 @@ func TestReaderCommittedStartOffset(t *testing.T) {
 			defer cleanup()
 
 			numMsgs := 10
-			msgs := make([]*proto.Message, numMsgs)
+			msgs := make([]*Message, numMsgs)
 			for i := 0; i < numMsgs; i++ {
-				msgs[i] = &proto.Message{
+				msgs[i] = &Message{
 					Value:       []byte(strconv.Itoa(i)),
 					Timestamp:   int64(i),
 					LeaderEpoch: 42,
@@ -223,11 +221,11 @@ func TestReaderCommittedReadError(t *testing.T) {
 	})
 	defer cleanup()
 
-	msg := &proto.Message{Value: []byte("hi")}
-	_, err := l.Append([]*proto.Message{msg})
+	msg := &Message{Value: []byte("hi")}
+	_, err := l.Append([]*Message{msg})
 	require.NoError(t, err)
-	msg = &proto.Message{Value: []byte("hi")}
-	_, err = l.Append([]*proto.Message{msg})
+	msg = &Message{Value: []byte("hi")}
+	_, err = l.Append([]*Message{msg})
 	require.NoError(t, err)
 	l.SetHighWatermark(0)
 
@@ -252,7 +250,7 @@ func TestReaderCommittedWaitOnEmptyLog(t *testing.T) {
 	r, err := l.NewReader(0, false)
 	require.NoError(t, err)
 
-	msg := &proto.Message{
+	msg := &Message{
 		Value:       []byte("hello"),
 		Timestamp:   1,
 		LeaderEpoch: 42,
@@ -260,7 +258,7 @@ func TestReaderCommittedWaitOnEmptyLog(t *testing.T) {
 
 	go func() {
 		time.Sleep(5 * time.Millisecond)
-		_, err := l.Append([]*proto.Message{msg})
+		_, err := l.Append([]*Message{msg})
 		require.NoError(t, err)
 		l.SetHighWatermark(0)
 	}()
@@ -286,9 +284,9 @@ func TestReaderCommittedRead(t *testing.T) {
 			defer cleanup()
 
 			numMsgs := 10
-			msgs := make([]*proto.Message, numMsgs)
+			msgs := make([]*Message, numMsgs)
 			for i := 0; i < numMsgs; i++ {
-				msgs[i] = &proto.Message{
+				msgs[i] = &Message{
 					Value:       []byte(strconv.Itoa(i)),
 					Timestamp:   int64(i),
 					LeaderEpoch: 42,
@@ -325,9 +323,9 @@ func TestReaderCommittedReadToHW(t *testing.T) {
 			defer cleanup()
 
 			numMsgs := 10
-			msgs := make([]*proto.Message, numMsgs)
+			msgs := make([]*Message, numMsgs)
 			for i := 0; i < numMsgs; i++ {
-				msgs[i] = &proto.Message{
+				msgs[i] = &Message{
 					Value:       []byte(strconv.Itoa(i)),
 					Timestamp:   int64(i),
 					LeaderEpoch: 42,
@@ -362,9 +360,9 @@ func TestReaderCommittedWaitForHW(t *testing.T) {
 	defer cleanup()
 
 	numMsgs := 10
-	msgs := make([]*proto.Message, numMsgs)
+	msgs := make([]*Message, numMsgs)
 	for i := 0; i < numMsgs; i++ {
-		msgs[i] = &proto.Message{
+		msgs[i] = &Message{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   int64(i),
 			LeaderEpoch: 42,
@@ -402,9 +400,9 @@ func TestReaderCommittedCancel(t *testing.T) {
 	defer cleanup()
 
 	numMsgs := 10
-	msgs := make([]*proto.Message, numMsgs)
+	msgs := make([]*Message, numMsgs)
 	for i := 0; i < numMsgs; i++ {
-		msgs[i] = &proto.Message{
+		msgs[i] = &Message{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   int64(i),
 			LeaderEpoch: 42,
@@ -449,19 +447,19 @@ func TestReaderCommittedCapOffset(t *testing.T) {
 	})
 	defer cleanup()
 
-	msg1 := &proto.Message{
+	msg1 := &Message{
 		Value:       []byte("hi"),
 		Timestamp:   1,
 		LeaderEpoch: 42,
 	}
-	_, err := l.Append([]*proto.Message{msg1})
+	_, err := l.Append([]*Message{msg1})
 	require.NoError(t, err)
-	msg2 := &proto.Message{
+	msg2 := &Message{
 		Value:       []byte("hi"),
 		Timestamp:   2,
 		LeaderEpoch: 42,
 	}
-	_, err = l.Append([]*proto.Message{msg2})
+	_, err = l.Append([]*Message{msg2})
 	require.NoError(t, err)
 	l.SetHighWatermark(0)
 
@@ -479,7 +477,7 @@ func TestReaderCommittedCapOffset(t *testing.T) {
 	compareMessages(t, msg2, m)
 }
 
-func compareMessages(t *testing.T, exp *proto.Message, act Message) {
+func compareMessages(t *testing.T, exp *Message, act SerializedMessage) {
 	// TODO: check timestamp
 	require.Equal(t, exp.MagicByte, act.MagicByte())
 	require.Equal(t, exp.Attributes, act.Attributes())
diff --git a/server/proto/envelope_test.go b/server/proto/envelope_test.go
new file mode 100644
index 0000000..aae4daf
--- /dev/null
+++ b/server/proto/envelope_test.go
@@ -0,0 +1,104 @@
+package proto
+
+import (
+	"testing"
+	"time"
+
+	client "github.com/liftbridge-io/liftbridge-api/go"
+	"github.com/stretchr/testify/require"
+)
+
+// Ensure we can marshal a message and then unmarshal it.
+func TestMarshalUnmarshalEnvelopeMessage(t *testing.T) {
+	msg := &client.Message{
+		Offset:       42,
+		Key:          []byte("foo"),
+		Value:        []byte("hello"),
+		Timestamp:    time.Now().UnixNano(),
+		Stream:       "foo",
+		Subject:      "foo",
+		ReplySubject: "reply",
+		Headers: map[string][]byte{
+			"foo": []byte("bar"),
+			"baz": []byte("qux"),
+		},
+		AckInbox:      "ack",
+		CorrelationId: "123",
+	}
+
+	envelope, err := MarshalEnvelope(msg)
+	require.NoError(t, err)
+
+	unmarshaled := new(client.Message)
+	require.NoError(t, UnmarshalEnvelope(envelope, unmarshaled))
+
+	require.Equal(t, msg, unmarshaled)
+}
+
+// Ensure we can marshal an ack and then unmarshal it.
+func TestMarshalUnmarshalEnvelopeAck(t *testing.T) {
+	ack := &client.Ack{
+		Offset:           42,
+		Stream:           "foo",
+		MsgSubject:       "foo",
+		AckInbox:         "ack",
+		CorrelationId:    "123",
+		PartitionSubject: "foo.1",
+	}
+
+	envelope, err := MarshalEnvelope(ack)
+	require.NoError(t, err)
+
+	unmarshaled := new(client.Ack)
+	require.NoError(t, UnmarshalEnvelope(envelope, unmarshaled))
+
+	require.Equal(t, ack, unmarshaled)
+}
+
+// Ensure UnmarshalEnvelope returns an error if there is not enough data for an
+// envelope.
+func TestUnmarshalEnvelopeUnderflow(t *testing.T) {
+	require.Error(t, UnmarshalEnvelope([]byte{}, new(client.Ack)))
+}
+
+// Ensure UnmarshalEnvelope returns an error if the magic number is different.
+func TestUnmarshalEnvelopeUnexpectedMagicNumber(t *testing.T) {
+	require.Error(t, UnmarshalEnvelope([]byte("foobarbaz"), new(client.Ack)))
+}
+
+// Ensure UnmarshalEnvelope returns an error if the protocol version is
+// unknown.
+func TestUnmarshalEnvelopeUnexpectedProtoVersion(t *testing.T) {
+	msg, err := MarshalEnvelope(new(client.Message))
+	require.NoError(t, err)
+	msg[4] = 0x01
+	require.Error(t, UnmarshalEnvelope(msg, new(client.Message)))
+}
+
+// Ensure UnmarshalEnvelope returns an error if the CRC flag is set but no CRC
+// is present.
+func TestUnmarshalEnvelopeMissingCRC(t *testing.T) {
+	msg, err := MarshalEnvelope(new(client.Message))
+	require.NoError(t, err)
+	msg[6] = setBit(msg[6], 0)
+	require.Error(t, UnmarshalEnvelope(msg, new(client.Message)))
+}
+
+// Ensure UnmarshalEnvelope returns an error if the CRC flag is set but the CRC
+// doesn't match the expected CRC.
+func TestUnmarshalEnvelopeMismatchedCRC(t *testing.T) {
+	msg, err := MarshalEnvelope(new(client.Message))
+	require.NoError(t, err)
+	msg[6] = setBit(msg[6], 0)
+	buf := make([]byte, len(msg)+4)
+	copy(buf, msg[:8])
+	buf[8] = byte(32)
+	copy(buf[12:], msg[8:])
+	buf[5] = byte(12)
+	require.Error(t, UnmarshalEnvelope(buf, new(client.Message)))
+}
+
+func setBit(n byte, pos uint8) byte {
+	n |= (1 << pos)
+	return n
+}
diff --git a/server/replicator_test.go b/server/replicator_test.go
index b501e5b..ce7af16 100644
--- a/server/replicator_test.go
+++ b/server/replicator_test.go
@@ -11,8 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	lift "github.com/liftbridge-io/go-liftbridge"
-	proto "github.com/liftbridge-io/liftbridge-api/go"
-	internal "github.com/liftbridge-io/liftbridge/server/proto"
+	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 func waitForHW(t *testing.T, timeout time.Duration, name string, partitionID int32, hw int64, servers ...*Server) {
@@ -129,7 +128,7 @@ func TestStreamLeaderFailover(t *testing.T) {
 
 	// Publish messages.
 	for i := 0; i < num; i++ {
-		_, err := client.Publish(context.Background(), subject, expected[i].Value,
+		_, err := client.Publish(context.Background(), name, expected[i].Value,
 			lift.Key(expected[i].Key), lift.AckPolicyAll())
 		require.NoError(t, err)
 	}
@@ -138,7 +137,7 @@ func TestStreamLeaderFailover(t *testing.T) {
 	i := 0
 	ch := make(chan struct{})
 	err = client.Subscribe(context.Background(), name,
-		func(msg *proto.Message, err error) {
+		func(msg lift.Message, err error) {
 			if i == num && err != nil {
 				return
 			}
@@ -179,7 +178,7 @@ func TestStreamLeaderFailover(t *testing.T) {
 	i = 0
 	ch = make(chan struct{})
 	err = client.Subscribe(context.Background(), name,
-		func(msg *proto.Message, err error) {
+		func(msg lift.Message, err error) {
 			if i == num && err != nil {
 				return
 			}
@@ -262,7 +261,7 @@ func TestCommitOnISRShrink(t *testing.T) {
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err := client.Publish(ctx, subject, []byte("hello"), lift.AckPolicyAll())
+		_, err := client.Publish(ctx, name, []byte("hello"), lift.AckPolicyAll())
 		gotAck <- err
 	}()
 
@@ -337,11 +336,10 @@ func TestAckPolicyLeader(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	cid := "cid"
-	ack, err := client.Publish(ctx, subject, []byte("hello"),
-		lift.CorrelationID(cid))
+	ack, err := client.Publish(ctx, name, []byte("hello"), lift.CorrelationID(cid))
 	require.NoError(t, err)
 	require.NotNil(t, ack)
-	require.Equal(t, cid, ack.CorrelationId)
+	require.Equal(t, cid, ack.CorrelationID())
 }
 
 // Ensure messages in the log still get committed after the leader is
@@ -384,7 +382,7 @@ func TestCommitOnRestart(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"), lift.AckPolicyAll())
+		_, err = client.Publish(ctx, name, []byte("hello"), lift.AckPolicyAll())
 		require.NoError(t, err)
 	}
 
@@ -404,7 +402,7 @@ func TestCommitOnRestart(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -436,12 +434,12 @@ func TestCommitOnRestart(t *testing.T) {
 	i := 0
 	ch := make(chan struct{})
 	err = client.Subscribe(context.Background(), name,
-		func(msg *proto.Message, err error) {
+		func(msg lift.Message, err error) {
 			if i == num*2 && err != nil {
 				return
 			}
 			require.NoError(t, err)
-			require.Equal(t, int64(i), msg.Offset)
+			require.Equal(t, int64(i), msg.Offset())
 			i++
 			if i == num*2 {
 				close(ch)
@@ -509,11 +507,11 @@ func TestTruncateFastLeaderElection(t *testing.T) {
 	// Publish two messages.
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	_, err = client.Publish(ctx, subject, []byte("hello"), lift.AckPolicyAll())
+	_, err = client.Publish(ctx, name, []byte("hello"), lift.AckPolicyAll())
 	require.NoError(t, err)
 	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	_, err = client.Publish(ctx, subject, []byte("world"), lift.AckPolicyAll())
+	_, err = client.Publish(ctx, name, []byte("world"), lift.AckPolicyAll())
 	require.NoError(t, err)
 
 	// Find stream followers.
@@ -643,11 +641,11 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	// Publish two messages.
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	_, err = client.Publish(ctx, subject, []byte("hello"))
+	_, err = client.Publish(ctx, name, []byte("hello"))
 	require.NoError(t, err)
 	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	_, err = client.Publish(ctx, subject, []byte("world"))
+	_, err = client.Publish(ctx, name, []byte("world"))
 	require.NoError(t, err)
 
 	// Find stream followers.
@@ -739,12 +737,12 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	// Publish new messages.
 	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	_, err = client.Publish(ctx, subject, []byte("goodnight"))
+	_, err = client.Publish(ctx, name, []byte("goodnight"))
 	require.NoError(t, err)
 
 	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	_, err = client.Publish(ctx, subject, []byte("moon"))
+	_, err = client.Publish(ctx, name, []byte("moon"))
 	require.NoError(t, err)
 
 	// Restart old leader.
@@ -836,12 +834,12 @@ func TestReplicatorNotifyNewData(t *testing.T) {
 	// aren't any messages. Set up a NATS subscription to intercept
 	// notifications.
 	var (
-		notifications = make(chan *internal.PartitionNotification)
+		notifications = make(chan *proto.PartitionNotification)
 		inbox         = follower.getPartitionNotificationInbox(
 			follower.config.Clustering.ServerID)
 	)
 	_, err = nc.Subscribe(inbox, func(msg *nats.Msg) {
-		req := &internal.PartitionNotification{}
+		req := &proto.PartitionNotification{}
 		if err := req.Unmarshal(msg.Data); err != nil {
 			t.Fatalf("Invalid partition notification: %v", err)
 		}
@@ -851,7 +849,7 @@ func TestReplicatorNotifyNewData(t *testing.T) {
 
 	// Publish a message. This will cause a notification to be sent to the
 	// follower.
-	_, err = client.Publish(context.Background(), subject, []byte("hello"))
+	_, err = client.Publish(context.Background(), name, []byte("hello"))
 	require.NoError(t, err)
 
 	select {
diff --git a/server/server_test.go b/server/server_test.go
index 9028b6b..36bfc29 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -18,8 +18,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health/grpc_health_v1"
 
-	proto "github.com/liftbridge-io/liftbridge-api/go"
-	internal "github.com/liftbridge-io/liftbridge/server/proto"
+	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 var storagePath string
@@ -514,7 +513,7 @@ func TestSubscribeOffsetOverflow(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -522,16 +521,16 @@ func TestSubscribeOffsetOverflow(t *testing.T) {
 	// starting at offset 5.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
-	err = client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
-		require.Equal(t, int64(5), msg.Offset)
+		require.Equal(t, int64(5), msg.Offset())
 		close(gotMsg)
 		cancel()
 	}, lift.StartAtOffset(100))
 	require.NoError(t, err)
 
 	// Publish one more message.
-	_, err = client.Publish(context.Background(), subject, []byte("test"))
+	_, err = client.Publish(context.Background(), name, []byte("test"))
 	require.NoError(t, err)
 
 	// Wait to get the new message.
@@ -573,16 +572,16 @@ func TestSubscribeOffsetOverflowEmptyStream(t *testing.T) {
 	// starting at offset 0.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
-	err = client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
-		require.Equal(t, int64(0), msg.Offset)
+		require.Equal(t, int64(0), msg.Offset())
 		close(gotMsg)
 		cancel()
 	})
 	require.NoError(t, err)
 
 	// Publish message.
-	_, err = client.Publish(context.Background(), subject, []byte("test"))
+	_, err = client.Publish(context.Background(), name, []byte("test"))
 	require.NoError(t, err)
 
 	// Wait to get the message.
@@ -629,7 +628,7 @@ func TestSubscribeOffsetUnderflow(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -639,9 +638,9 @@ func TestSubscribeOffsetUnderflow(t *testing.T) {
 	// Subscribe with underflowed offset. This should set the offset to 1.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
-	err = client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
-		require.Equal(t, int64(1), msg.Offset)
+		require.Equal(t, int64(1), msg.Offset())
 		close(gotMsg)
 		cancel()
 	}, lift.StartAtOffset(0))
@@ -690,7 +689,7 @@ func TestStreamRetentionBytes(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -698,9 +697,9 @@ func TestStreamRetentionBytes(t *testing.T) {
 	forceLogClean(t, subject, name, s1)
 
 	// The first message read back should have offset 87.
-	msgs := make(chan *proto.Message, 1)
+	msgs := make(chan lift.Message, 1)
 	ctx, cancel := context.WithCancel(context.Background())
-	err = client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
 		msgs <- msg
 		cancel()
@@ -710,7 +709,7 @@ func TestStreamRetentionBytes(t *testing.T) {
 	// Wait to get the new message.
 	select {
 	case msg := <-msgs:
-		require.Equal(t, int64(87), msg.Offset)
+		require.Equal(t, int64(87), msg.Offset())
 	case <-time.After(5 * time.Second):
 		t.Fatal("Did not receive expected message")
 	}
@@ -751,7 +750,7 @@ func TestStreamRetentionMessages(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -759,9 +758,9 @@ func TestStreamRetentionMessages(t *testing.T) {
 	forceLogClean(t, subject, name, s1)
 
 	// The first message read back should have offset 5.
-	msgs := make(chan *proto.Message, 1)
+	msgs := make(chan lift.Message, 1)
 	ctx, cancel := context.WithCancel(context.Background())
-	err = client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
 		msgs <- msg
 		cancel()
@@ -771,7 +770,7 @@ func TestStreamRetentionMessages(t *testing.T) {
 	// Wait to get the new message.
 	select {
 	case msg := <-msgs:
-		require.Equal(t, int64(5), msg.Offset)
+		require.Equal(t, int64(5), msg.Offset())
 	case <-time.After(5 * time.Second):
 		t.Fatal("Did not receive expected message")
 	}
@@ -812,7 +811,7 @@ func TestStreamRetentionAge(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -821,9 +820,9 @@ func TestStreamRetentionAge(t *testing.T) {
 
 	// We expect all segments but the last to be truncated due to age, so the
 	// first message read back should have offset 99.
-	msgs := make(chan *proto.Message, 1)
+	msgs := make(chan lift.Message, 1)
 	ctx, cancel := context.WithCancel(context.Background())
-	err = client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
 		msgs <- msg
 		cancel()
@@ -833,7 +832,7 @@ func TestStreamRetentionAge(t *testing.T) {
 	// Wait to get the new message.
 	select {
 	case msg := <-msgs:
-		require.Equal(t, int64(99), msg.Offset)
+		require.Equal(t, int64(99), msg.Offset())
 	case <-time.After(5 * time.Second):
 		t.Fatal("Did not receive expected message")
 	}
@@ -875,7 +874,7 @@ func TestSubscribeEarliest(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -885,9 +884,9 @@ func TestSubscribeEarliest(t *testing.T) {
 	// Subscribe with EARLIEST. This should start reading from offset 1.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
-	client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
-		require.Equal(t, int64(1), msg.Offset)
+		require.Equal(t, int64(1), msg.Offset())
 		close(gotMsg)
 		cancel()
 	}, lift.StartAtEarliestReceived())
@@ -932,16 +931,16 @@ func TestSubscribeLatest(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
 	// Subscribe with LATEST. This should start reading from offset 2.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
-	client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
-		require.Equal(t, int64(2), msg.Offset)
+		require.Equal(t, int64(2), msg.Offset())
 		close(gotMsg)
 		cancel()
 	}, lift.StartAtLatestReceived())
@@ -986,7 +985,7 @@ func TestSubscribeNewOnly(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
@@ -994,16 +993,16 @@ func TestSubscribeNewOnly(t *testing.T) {
 	// offset 5.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
-	err = client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	err = client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		require.NoError(t, err)
-		require.Equal(t, int64(5), msg.Offset)
+		require.Equal(t, int64(5), msg.Offset())
 		close(gotMsg)
 		cancel()
 	})
 	require.NoError(t, err)
 
 	// Publish one more message.
-	_, err = client.Publish(context.Background(), subject, []byte("test"))
+	_, err = client.Publish(context.Background(), name, []byte("test"))
 	require.NoError(t, err)
 
 	// Wait to get the new message.
@@ -1056,22 +1055,22 @@ func TestSubscribeStartTime(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_, err = client.Publish(ctx, subject, []byte("hello"))
+		_, err = client.Publish(ctx, name, []byte("hello"))
 		require.NoError(t, err)
 	}
 
 	// Subscribe with TIMESTAMP 25. This should start reading from offset 3.
 	gotMsg := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
-	client.Subscribe(ctx, name, func(msg *proto.Message, err error) {
+	client.Subscribe(ctx, name, func(msg lift.Message, err error) {
 		select {
 		case <-gotMsg:
 			return
 		default:
 		}
 		require.NoError(t, err)
-		require.Equal(t, int64(3), msg.Offset)
-		require.Equal(t, int64(30), msg.Timestamp)
+		require.Equal(t, int64(3), msg.Offset())
+		require.Equal(t, int64(30), msg.Timestamp().UnixNano())
 		close(gotMsg)
 		cancel()
 	}, lift.StartAtTime(time.Unix(0, 25)))
@@ -1293,8 +1292,8 @@ func TestPropagatedShrinkExpandISR(t *testing.T) {
 	leaderEpoch := partition.LeaderEpoch
 
 	// Shrink ISR.
-	controller.handleShrinkISR(&internal.PropagatedRequest{
-		ShrinkISROp: &internal.ShrinkISROp{
+	controller.handleShrinkISR(&proto.PropagatedRequest{
+		ShrinkISROp: &proto.ShrinkISROp{
 			Stream:          name,
 			Partition:       0,
 			ReplicaToRemove: followerID,
@@ -1307,9 +1306,9 @@ func TestPropagatedShrinkExpandISR(t *testing.T) {
 	waitForISR(t, 10*time.Second, name, 0, 1, s1, s2)
 
 	// Expand ISR.
-	controller.handleExpandISR(&internal.PropagatedRequest{
-		Op: internal.Op_EXPAND_ISR,
-		ExpandISROp: &internal.ExpandISROp{
+	controller.handleExpandISR(&proto.PropagatedRequest{
+		Op: proto.Op_EXPAND_ISR,
+		ExpandISROp: &proto.ExpandISROp{
 			Stream:       name,
 			Partition:    0,
 			ReplicaToAdd: followerID,