diff --git a/app/child_tx_decoder.go b/app/child_tx_decoder.go
index 3f3b8cdec0..768d0b1a32 100644
--- a/app/child_tx_decoder.go
+++ b/app/child_tx_decoder.go
@@ -7,8 +7,8 @@ import (
 
 func MalleatedTxDecoder(dec sdk.TxDecoder) sdk.TxDecoder {
 	return func(txBytes []byte) (sdk.Tx, error) {
-		if _, childTx, has := coretypes.UnwrapMalleatedTx(txBytes); has {
-			return dec(childTx)
+		if malleatedTx, has := coretypes.UnwrapMalleatedTx(txBytes); has {
+			return dec(malleatedTx.Tx)
 		}
 		return dec(txBytes)
 	}
diff --git a/app/process_proposal.go b/app/process_proposal.go
index 4ecc1283ff..47c2c9b2ee 100644
--- a/app/process_proposal.go
+++ b/app/process_proposal.go
@@ -121,7 +121,7 @@ func (app *App) ProcessProposal(req abci.RequestProcessProposal) abci.ResponsePr
 		}
 	}
 
-	nsshares, _, err := shares.ComputeShares(&data, req.BlockData.OriginalSquareSize)
+	rawShares, err := shares.Split(data)
 	if err != nil {
 		app.Logger().Error(rejectedPropBlockLog, "reason", "failure to compute shares from block data:", "error", err, "proposerAddress", req.Header.ProposerAddress)
 		return abci.ResponseProcessProposal{
@@ -129,7 +129,7 @@ func (app *App) ProcessProposal(req abci.RequestProcessProposal) abci.ResponsePr
 		}
 	}
 
-	eds, err := da.ExtendShares(req.BlockData.OriginalSquareSize, nsshares.RawShares())
+	eds, err := da.ExtendShares(req.BlockData.OriginalSquareSize, rawShares)
 	if err != nil {
 		app.Logger().Error(
 			rejectedPropBlockLog,
diff --git a/app/split_shares.go b/app/split_shares.go
index 1750c5b2c7..53d0532b5d 100644
--- a/app/split_shares.go
+++ b/app/split_shares.go
@@ -137,7 +137,10 @@ func newShareSplitter(txConf client.TxConfig, squareSize uint64, data *core.Data
 	if err != nil {
 		panic(err)
 	}
-	sqwr.evdShares = shares.SplitEvidenceIntoShares(evdData).RawShares()
+	sqwr.evdShares, err = shares.SplitEvidence(evdData.Evidence)
+	if err != nil {
+		panic(err)
+	}
 
 	sqwr.txWriter = coretypes.NewContiguousShareWriter(consts.TxNamespaceID)
 	sqwr.msgWriter = coretypes.NewMessageShareWriter()
@@ -188,7 +191,9 @@ func (sqwr *shareSplitter) writeMalleatedTx(
 		return false, nil, nil, err
 	}
 
-	wrappedTx, err := coretypes.WrapMalleatedTx(parentHash[:], rawProcessedTx)
+	// we use a share index of 0 here because this implementation doesn't
+	// support non-interactive defaults or the usuage of wrapped txs
+	wrappedTx, err := coretypes.WrapMalleatedTx(parentHash[:], 0, rawProcessedTx)
 	if err != nil {
 		return false, nil, nil, err
 	}
diff --git a/app/test/process_proposal_test.go b/app/test/process_proposal_test.go
index ef27875930..db816ad189 100644
--- a/app/test/process_proposal_test.go
+++ b/app/test/process_proposal_test.go
@@ -151,10 +151,10 @@ func TestMessageInclusionCheck(t *testing.T) {
 		data, err := coretypes.DataFromProto(tt.input.BlockData)
 		require.NoError(t, err)
 
-		shares, _, err := shares.ComputeShares(&data, tt.input.BlockData.OriginalSquareSize)
+		shares, err := shares.Split(data)
 		require.NoError(t, err)
 
-		rawShares := shares.RawShares()
+		rawShares := shares
 
 		require.NoError(t, err)
 		eds, err := da.ExtendShares(tt.input.BlockData.OriginalSquareSize, rawShares)
@@ -207,13 +207,11 @@ func TestProcessMessagesWithReservedNamespaces(t *testing.T) {
 		data, err := coretypes.DataFromProto(input.BlockData)
 		require.NoError(t, err)
 
-		shares, _, err := shares.ComputeShares(&data, input.BlockData.OriginalSquareSize)
+		shares, err := shares.Split(data)
 		require.NoError(t, err)
 
-		rawShares := shares.RawShares()
-
 		require.NoError(t, err)
-		eds, err := da.ExtendShares(input.BlockData.OriginalSquareSize, rawShares)
+		eds, err := da.ExtendShares(input.BlockData.OriginalSquareSize, shares)
 		require.NoError(t, err)
 		dah := da.NewDataAvailabilityHeader(eds)
 		input.Header.DataHash = dah.Hash()
@@ -234,6 +232,9 @@ func TestProcessMessageWithUnsortedMessages(t *testing.T) {
 	pfdOne, msgOne := genRandMsgPayForDataForNamespace(t, signer, 8, namespaceOne)
 	pfdTwo, msgTwo := genRandMsgPayForDataForNamespace(t, signer, 8, namespaceTwo)
 
+	cMsgOne := &core.Message{NamespaceId: pfdOne.GetMessageNamespaceId(), Data: msgOne}
+	cMsgTwo := &core.Message{NamespaceId: pfdTwo.GetMessageNamespaceId(), Data: msgTwo}
+
 	input := abci.RequestProcessProposal{
 		BlockData: &core.Data{
 			Txs: [][]byte{
@@ -242,14 +243,8 @@ func TestProcessMessageWithUnsortedMessages(t *testing.T) {
 			},
 			Messages: core.Messages{
 				MessagesList: []*core.Message{
-					{
-						NamespaceId: pfdTwo.GetMessageNamespaceId(),
-						Data:        msgTwo,
-					},
-					{
-						NamespaceId: pfdOne.GetMessageNamespaceId(),
-						Data:        msgOne,
-					},
+					cMsgOne,
+					cMsgTwo,
 				},
 			},
 			OriginalSquareSize: 8,
@@ -258,17 +253,20 @@ func TestProcessMessageWithUnsortedMessages(t *testing.T) {
 	data, err := coretypes.DataFromProto(input.BlockData)
 	require.NoError(t, err)
 
-	shares, _, err := shares.ComputeShares(&data, input.BlockData.OriginalSquareSize)
+	shares, err := shares.Split(data)
 	require.NoError(t, err)
 
-	rawShares := shares.RawShares()
-
 	require.NoError(t, err)
-	eds, err := da.ExtendShares(input.BlockData.OriginalSquareSize, rawShares)
+	eds, err := da.ExtendShares(input.BlockData.OriginalSquareSize, shares)
+
 	require.NoError(t, err)
 	dah := da.NewDataAvailabilityHeader(eds)
 	input.Header.DataHash = dah.Hash()
 
+	// swap the messages
+	input.BlockData.Messages.MessagesList[0] = cMsgTwo
+	input.BlockData.Messages.MessagesList[1] = cMsgOne
+
 	got := testApp.ProcessProposal(input)
 
 	assert.Equal(t, got.Result, abci.ResponseProcessProposal_REJECT)
diff --git a/app/test/split_shares_test.go b/app/test/split_shares_test.go
index 46e8a4e786..6962425ed3 100644
--- a/app/test/split_shares_test.go
+++ b/app/test/split_shares_test.go
@@ -13,7 +13,6 @@ import (
 	"github.com/tendermint/tendermint/pkg/consts"
 	"github.com/tendermint/tendermint/pkg/da"
 	core "github.com/tendermint/tendermint/proto/tendermint/types"
-	coretypes "github.com/tendermint/tendermint/types"
 )
 
 func TestSplitShares(t *testing.T) {
@@ -101,14 +100,14 @@ func TestSplitShares(t *testing.T) {
 		dah := da.NewDataAvailabilityHeader(eds)
 		data.Hash = dah.Hash()
 
-		parsedData, err := coretypes.DataFromSquare(eds)
+		parsedData, err := shares.Merge(eds)
 		require.NoError(t, err)
 
 		assert.Equal(t, data.Txs, parsedData.Txs.ToSliceOfBytes())
 
-		parsedShares, _, err := shares.ComputeShares(&parsedData, tt.squareSize)
+		parsedShares, err := shares.Split(parsedData)
 		require.NoError(t, err)
 
-		require.Equal(t, square, parsedShares.RawShares())
+		require.Equal(t, square, parsedShares)
 	}
 }
diff --git a/go.mod b/go.mod
index 8e97d15a6f..a5e060b9d5 100644
--- a/go.mod
+++ b/go.mod
@@ -161,5 +161,5 @@ require (
 replace (
 	github.com/cosmos/cosmos-sdk => github.com/celestiaorg/cosmos-sdk v1.2.0-sdk-v0.46.0
 	github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1
-	github.com/tendermint/tendermint => github.com/celestiaorg/celestia-core v1.3.3-tm-v0.34.20
+	github.com/tendermint/tendermint => github.com/celestiaorg/celestia-core v1.3.5-tm-v0.34.20
 )
diff --git a/go.sum b/go.sum
index 99bb6014ac..6f86ef3635 100644
--- a/go.sum
+++ b/go.sum
@@ -173,8 +173,8 @@ github.com/btcsuite/winsvc v1.0.0/go.mod h1:jsenWakMcC0zFBFurPLEAyrnc/teJEM1O46f
 github.com/bwesterb/go-ristretto v1.2.0/go.mod h1:fUIoIZaG73pV5biE2Blr2xEzDoMj7NFEuV9ekS419A0=
 github.com/c-bata/go-prompt v0.2.2/go.mod h1:VzqtzE2ksDBcdln8G7mk2RX9QyGjH+OVqOCSiVIqS34=
 github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=
-github.com/celestiaorg/celestia-core v1.3.3-tm-v0.34.20 h1:XspZtkoIfJ68TCVOOGWkI4W7taQFJLLqSu6GME5RbqU=
-github.com/celestiaorg/celestia-core v1.3.3-tm-v0.34.20/go.mod h1:f4R8qNJrP1CDH0SNwj4jA3NymBLQM4lNdx6Ijmfllbw=
+github.com/celestiaorg/celestia-core v1.3.5-tm-v0.34.20 h1:Z66gewBIJIQ5T72v2ktu4FMA/sgDE4ok4y9DvfnCUUE=
+github.com/celestiaorg/celestia-core v1.3.5-tm-v0.34.20/go.mod h1:f4R8qNJrP1CDH0SNwj4jA3NymBLQM4lNdx6Ijmfllbw=
 github.com/celestiaorg/cosmos-sdk v1.2.0-sdk-v0.46.0 h1:A1F7L/09uGClsU+kmugMy47Ezv4ll0uxNRNdaGa37T8=
 github.com/celestiaorg/cosmos-sdk v1.2.0-sdk-v0.46.0/go.mod h1:OXRC0p460CFKl77uQZWY/8p5uZmDrNum7BmVZDupq0Q=
 github.com/celestiaorg/go-leopard v0.1.0 h1:28z2EkvKJIez5J9CEaiiUEC+OxalRLtTGJJ1oScfE1g=
diff --git a/pkg/appconsts/appconsts.go b/pkg/appconsts/appconsts.go
new file mode 100644
index 0000000000..1a104ff3b0
--- /dev/null
+++ b/pkg/appconsts/appconsts.go
@@ -0,0 +1,9 @@
+package appconsts
+
+import (
+	"bytes"
+
+	"github.com/tendermint/tendermint/pkg/consts"
+)
+
+var NameSpacedPaddedShareBytes = bytes.Repeat([]byte{0}, consts.MsgShareSize)
diff --git a/pkg/shares/contiguous_shares_test.go b/pkg/shares/contiguous_shares_test.go
new file mode 100644
index 0000000000..2bc3b74a88
--- /dev/null
+++ b/pkg/shares/contiguous_shares_test.go
@@ -0,0 +1,123 @@
+package shares
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/tendermint/tendermint/pkg/consts"
+	coretypes "github.com/tendermint/tendermint/types"
+)
+
+func TestContigShareWriter(t *testing.T) {
+	// note that this test is mainly for debugging purposes, the main round trip
+	// tests occur in TestMerge and Test_processContiguousShares
+	w := NewContiguousShareSplitter(consts.TxNamespaceID)
+	txs := generateRandomContiguousShares(33, 200)
+	for _, tx := range txs {
+		rawTx, _ := tx.MarshalDelimited()
+		w.WriteBytes(rawTx)
+	}
+	resShares := w.Export()
+	rawResTxs, err := processContiguousShares(resShares.RawShares())
+	resTxs := coretypes.ToTxs(rawResTxs)
+	require.NoError(t, err)
+
+	assert.Equal(t, txs, resTxs)
+}
+
+func Test_parseDelimiter(t *testing.T) {
+	for i := uint64(0); i < 100; i++ {
+		tx := generateRandomContiguousShares(1, int(i))[0]
+		input, err := tx.MarshalDelimited()
+		if err != nil {
+			panic(err)
+		}
+		res, txLen, err := ParseDelimiter(input)
+		if err != nil {
+			panic(err)
+		}
+		assert.Equal(t, i, txLen)
+		assert.Equal(t, []byte(tx), res)
+	}
+}
+
+func TestFuzz_processContiguousShares(t *testing.T) {
+	t.Skip()
+	// run random shares through processContiguousShares for a minute
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			Test_processContiguousShares(t)
+		}
+	}
+}
+
+func Test_processContiguousShares(t *testing.T) {
+	// exactTxShareSize is the length of tx that will fit exactly into a single
+	// share, accounting for namespace id and the length delimiter prepended to
+	// each tx
+	const exactTxShareSize = consts.TxShareSize - 1
+
+	type test struct {
+		name    string
+		txSize  int
+		txCount int
+	}
+
+	// each test is ran twice, once using txSize as an exact size, and again
+	// using it as a cap for randomly sized txs
+	tests := []test{
+		{"single small tx", 10, 1},
+		{"many small txs", 10, 10},
+		{"single big tx", 1000, 1},
+		{"many big txs", 1000, 10},
+		{"single exact size tx", exactTxShareSize, 1},
+		{"many exact size txs", exactTxShareSize, 10},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+
+		// run the tests with identically sized txs
+		t.Run(fmt.Sprintf("%s idendically sized ", tc.name), func(t *testing.T) {
+			txs := generateRandomContiguousShares(tc.txCount, tc.txSize)
+
+			shares := SplitTxs(txs)
+
+			parsedTxs, err := processContiguousShares(shares)
+			if err != nil {
+				t.Error(err)
+			}
+
+			// check that the data parsed is identical
+			for i := 0; i < len(txs); i++ {
+				assert.Equal(t, []byte(txs[i]), parsedTxs[i])
+			}
+		})
+
+		// run the same tests using randomly sized txs with caps of tc.txSize
+		t.Run(fmt.Sprintf("%s randomly sized", tc.name), func(t *testing.T) {
+			txs := generateRandomlySizedContiguousShares(tc.txCount, tc.txSize)
+
+			shares := SplitTxs(txs)
+
+			parsedTxs, err := processContiguousShares(shares)
+			if err != nil {
+				t.Error(err)
+			}
+
+			// check that the data parsed is identical to the original
+			for i := 0; i < len(txs); i++ {
+				assert.Equal(t, []byte(txs[i]), parsedTxs[i])
+			}
+		})
+	}
+}
diff --git a/pkg/shares/message_shares_test.go b/pkg/shares/message_shares_test.go
new file mode 100644
index 0000000000..db4e9485d9
--- /dev/null
+++ b/pkg/shares/message_shares_test.go
@@ -0,0 +1,99 @@
+package shares
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/tendermint/tendermint/pkg/consts"
+	coretypes "github.com/tendermint/tendermint/types"
+)
+
+func Test_parseMsgShares(t *testing.T) {
+	// exactMsgShareSize is the length of message that will fit exactly into a single
+	// share, accounting for namespace id and the length delimiter prepended to
+	// each message
+	const exactMsgShareSize = consts.MsgShareSize - 2
+
+	type test struct {
+		name     string
+		msgSize  int
+		msgCount int
+	}
+
+	// each test is ran twice, once using msgSize as an exact size, and again
+	// using it as a cap for randomly sized leaves
+	tests := []test{
+		{"single small msg", 100, 1},
+		{"many small msgs", 100, 10},
+		{"single big msg", 1000, 1},
+		{"many big msgs", 1000, 10},
+		{"single exact size msg", exactMsgShareSize, 1},
+		{"many exact size msgs", consts.MsgShareSize, 10},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		// run the tests with identically sized messages
+		t.Run(fmt.Sprintf("%s identically sized ", tc.name), func(t *testing.T) {
+			rawmsgs := make([]coretypes.Message, tc.msgCount)
+			for i := 0; i < tc.msgCount; i++ {
+				rawmsgs[i] = generateRandomMessage(tc.msgSize)
+			}
+
+			msgs := coretypes.Messages{MessagesList: rawmsgs}
+			msgs.SortMessages()
+
+			shares, _ := SplitMessages(nil, msgs.MessagesList)
+
+			parsedMsgs, err := parseMsgShares(shares)
+			if err != nil {
+				t.Error(err)
+			}
+
+			// check that the namespaces and data are the same
+			for i := 0; i < len(msgs.MessagesList); i++ {
+				assert.Equal(t, msgs.MessagesList[i].NamespaceID, parsedMsgs[i].NamespaceID)
+				assert.Equal(t, msgs.MessagesList[i].Data, parsedMsgs[i].Data)
+			}
+		})
+
+		// run the same tests using randomly sized messages with caps of tc.msgSize
+		t.Run(fmt.Sprintf("%s randomly sized", tc.name), func(t *testing.T) {
+			msgs := generateRandomlySizedMessages(tc.msgCount, tc.msgSize)
+			shares, _ := SplitMessages(nil, msgs.MessagesList)
+
+			parsedMsgs, err := parseMsgShares(shares)
+			if err != nil {
+				t.Error(err)
+			}
+
+			// check that the namesapces and data are the same
+			for i := 0; i < len(msgs.MessagesList); i++ {
+				assert.Equal(t, msgs.MessagesList[i].NamespaceID, parsedMsgs[i].NamespaceID)
+				assert.Equal(t, msgs.MessagesList[i].Data, parsedMsgs[i].Data)
+			}
+		})
+	}
+}
+
+func TestParsePaddedMsg(t *testing.T) {
+	msgWr := NewMessageShareSplitter()
+	randomSmallMsg := generateRandomMessage(100)
+	randomLargeMsg := generateRandomMessage(10000)
+	msgs := coretypes.Messages{
+		MessagesList: []coretypes.Message{
+			randomSmallMsg,
+			randomLargeMsg,
+		},
+	}
+	msgs.SortMessages()
+	msgWr.Write(msgs.MessagesList[0])
+	msgWr.WriteNamespacedPaddedShares(4)
+	msgWr.Write(msgs.MessagesList[1])
+	msgWr.WriteNamespacedPaddedShares(10)
+	pmsgs, err := parseMsgShares(msgWr.Export().RawShares())
+	require.NoError(t, err)
+	require.Equal(t, msgs.MessagesList, pmsgs)
+}
diff --git a/pkg/shares/parse_contiguous_shares.go b/pkg/shares/parse_contiguous_shares.go
new file mode 100644
index 0000000000..450d60b795
--- /dev/null
+++ b/pkg/shares/parse_contiguous_shares.go
@@ -0,0 +1,83 @@
+package shares
+
+import (
+	"encoding/binary"
+	"errors"
+
+	"github.com/tendermint/tendermint/pkg/consts"
+)
+
+// processContiguousShares takes raw shares and extracts out transactions,
+// intermediate state roots, or evidence. The returned [][]byte do not have
+// namespaces or length delimiters and are ready to be unmarshalled
+func processContiguousShares(shares [][]byte) (txs [][]byte, err error) {
+	if len(shares) == 0 {
+		return nil, nil
+	}
+
+	ss := newShareStack(shares)
+	return ss.resolve()
+}
+
+// shareStack holds variables for peel
+type shareStack struct {
+	shares  [][]byte
+	dataLen uint64
+	// data may be transactions, intermediate state roots, or evidence depending
+	// on the namespace ID for this share
+	data   [][]byte
+	cursor int
+}
+
+func newShareStack(shares [][]byte) *shareStack {
+	return &shareStack{shares: shares}
+}
+
+func (ss *shareStack) resolve() ([][]byte, error) {
+	if len(ss.shares) == 0 {
+		return nil, nil
+	}
+	err := ss.peel(ss.shares[0][consts.NamespaceSize+consts.ShareReservedBytes:], true)
+	return ss.data, err
+}
+
+// peel recursively parses each chunk of data (either a transaction,
+// intermediate state root, or evidence) and adds it to the underlying slice of data.
+func (ss *shareStack) peel(share []byte, delimited bool) (err error) {
+	if delimited {
+		var txLen uint64
+		share, txLen, err = ParseDelimiter(share)
+		if err != nil {
+			return err
+		}
+		if txLen == 0 {
+			return nil
+		}
+		ss.dataLen = txLen
+	}
+	// safeLen describes the point in the share where it can be safely split. If
+	// split beyond this point, it is possible to break apart a length
+	// delimiter, which will result in incorrect share merging
+	safeLen := len(share) - binary.MaxVarintLen64
+	if safeLen < 0 {
+		safeLen = 0
+	}
+	if ss.dataLen <= uint64(safeLen) {
+		ss.data = append(ss.data, share[:ss.dataLen])
+		share = share[ss.dataLen:]
+		return ss.peel(share, true)
+	}
+	// add the next share to the current share to continue merging if possible
+	if len(ss.shares) > ss.cursor+1 {
+		ss.cursor++
+		share := append(share, ss.shares[ss.cursor][consts.NamespaceSize+consts.ShareReservedBytes:]...)
+		return ss.peel(share, false)
+	}
+	// collect any remaining data
+	if ss.dataLen <= uint64(len(share)) {
+		ss.data = append(ss.data, share[:ss.dataLen])
+		share = share[ss.dataLen:]
+		return ss.peel(share, true)
+	}
+	return errors.New("failure to parse block data: transaction length exceeded data length")
+}
diff --git a/pkg/shares/parse_message_shares.go b/pkg/shares/parse_message_shares.go
new file mode 100644
index 0000000000..cfc0ccfa2f
--- /dev/null
+++ b/pkg/shares/parse_message_shares.go
@@ -0,0 +1,114 @@
+package shares
+
+import (
+	"bytes"
+	"encoding/binary"
+
+	"github.com/celestiaorg/celestia-app/pkg/appconsts"
+	"github.com/tendermint/tendermint/pkg/consts"
+	coretypes "github.com/tendermint/tendermint/types"
+)
+
+// parseMsgShares iterates through raw shares and separates the contiguous chunks
+// of data. It is only used for Messages, i.e. shares with a non-reserved namespace.
+func parseMsgShares(shares [][]byte) ([]coretypes.Message, error) {
+	if len(shares) == 0 {
+		return nil, nil
+	}
+	// msgs returned
+	msgs := []coretypes.Message{}
+	currentMsgLen := 0
+	currentMsg := coretypes.Message{}
+	// whether the current share contains the start of a new message
+	isNewMessage := true
+	// the len in bytes of the current chunk of data that will eventually become
+	// a message. This is identical to len(currentMsg.Data) + consts.MsgShareSize
+	// but we cache it here for readability
+	dataLen := 0
+	saveMessage := func() {
+		msgs = append(msgs, currentMsg)
+		dataLen = 0
+		isNewMessage = true
+	}
+	// iterate through all the shares and parse out each msg
+	for i := 0; i < len(shares); i++ {
+		dataLen = len(currentMsg.Data) + consts.MsgShareSize
+		switch {
+		case isNewMessage:
+			nextMsgChunk, nextMsgLen, err := ParseDelimiter(shares[i][consts.NamespaceSize:])
+			if err != nil {
+				return nil, err
+			}
+			// the current share is namespaced padding so we ignore it
+			if bytes.Equal(shares[i][consts.NamespaceSize:], appconsts.NameSpacedPaddedShareBytes) {
+				continue
+			}
+			currentMsgLen = int(nextMsgLen)
+			nid := shares[i][:consts.NamespaceSize]
+			currentMsg = coretypes.Message{
+				NamespaceID: nid,
+				Data:        nextMsgChunk,
+			}
+			// the current share contains the entire msg so we save it and
+			// progress
+			if currentMsgLen <= len(nextMsgChunk) {
+				currentMsg.Data = currentMsg.Data[:currentMsgLen]
+				saveMessage()
+				continue
+			}
+			isNewMessage = false
+		// this entire share contains a chunk of message that we need to save
+		case currentMsgLen > dataLen:
+			currentMsg.Data = append(currentMsg.Data, shares[i][consts.NamespaceSize:]...)
+		// this share contains the last chunk of data needed to complete the
+		// message
+		case currentMsgLen <= dataLen:
+			remaining := currentMsgLen - len(currentMsg.Data) + consts.NamespaceSize
+			currentMsg.Data = append(currentMsg.Data, shares[i][consts.NamespaceSize:remaining]...)
+			saveMessage()
+		}
+	}
+	return msgs, nil
+}
+
+// ParseDelimiter finds and returns the length delimiter of the message provided
+// while also removing the delimiter bytes from the input
+func ParseDelimiter(input []byte) ([]byte, uint64, error) {
+	if len(input) == 0 {
+		return input, 0, nil
+	}
+
+	l := binary.MaxVarintLen64
+	if len(input) < binary.MaxVarintLen64 {
+		l = len(input)
+	}
+
+	delimiter := zeroPadIfNecessary(input[:l], binary.MaxVarintLen64)
+
+	// read the length of the message
+	r := bytes.NewBuffer(delimiter)
+	msgLen, err := binary.ReadUvarint(r)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	// calculate the number of bytes used by the delimiter
+	lenBuf := make([]byte, binary.MaxVarintLen64)
+	n := binary.PutUvarint(lenBuf, msgLen)
+
+	// return the input without the length delimiter
+	return input[n:], msgLen, nil
+}
+
+func zeroPadIfNecessary(share []byte, width int) []byte {
+	oldLen := len(share)
+	if oldLen >= width {
+		return share
+	}
+
+	missingBytes := width - oldLen
+	padByte := []byte{0}
+	padding := bytes.Repeat(padByte, missingBytes)
+	share = append(share, padding...)
+	return share
+}
diff --git a/pkg/shares/share_merging.go b/pkg/shares/share_merging.go
index da6fdf3a03..a7c247e65e 100644
--- a/pkg/shares/share_merging.go
+++ b/pkg/shares/share_merging.go
@@ -1,9 +1,7 @@
-package types
+package shares
 
 import (
 	"bytes"
-	"encoding/binary"
-	"errors"
 
 	"github.com/celestiaorg/rsmt2d"
 	"github.com/gogo/protobuf/proto"
@@ -12,8 +10,8 @@ import (
 	coretypes "github.com/tendermint/tendermint/types"
 )
 
-// DataFromSquare extracts block data from an extended data square.
-func DataFromSquare(eds *rsmt2d.ExtendedDataSquare) (coretypes.Data, error) {
+// Merge extracts block data from an extended data square.
+func Merge(eds *rsmt2d.ExtendedDataSquare) (coretypes.Data, error) {
 	originalWidth := eds.Width() / 2
 
 	// sort block data shares by namespace
@@ -69,9 +67,10 @@ func DataFromSquare(eds *rsmt2d.ExtendedDataSquare) (coretypes.Data, error) {
 	}
 
 	return coretypes.Data{
-		Txs:      txs,
-		Evidence: evd,
-		Messages: msgs,
+		Txs:                txs,
+		Evidence:           evd,
+		Messages:           msgs,
+		OriginalSquareSize: uint64(originalWidth),
 	}, nil
 }
 
@@ -133,176 +132,3 @@ func ParseMsgs(shares [][]byte) (coretypes.Messages, error) {
 		MessagesList: msgList,
 	}, nil
 }
-
-// processContiguousShares takes raw shares and extracts out transactions,
-// intermediate state roots, or evidence. The returned [][]byte do have
-// namespaces or length delimiters and are ready to be unmarshalled
-func processContiguousShares(shares [][]byte) (txs [][]byte, err error) {
-	if len(shares) == 0 {
-		return nil, nil
-	}
-
-	ss := newShareStack(shares)
-	return ss.resolve()
-}
-
-// shareStack hold variables for peel
-type shareStack struct {
-	shares [][]byte
-	txLen  uint64
-	txs    [][]byte
-	cursor int
-}
-
-func newShareStack(shares [][]byte) *shareStack {
-	return &shareStack{shares: shares}
-}
-
-func (ss *shareStack) resolve() ([][]byte, error) {
-	if len(ss.shares) == 0 {
-		return nil, nil
-	}
-	err := ss.peel(ss.shares[0][consts.NamespaceSize+consts.ShareReservedBytes:], true)
-	return ss.txs, err
-}
-
-// peel recursively parses each chunk of data (either a transaction,
-// intermediate state root, or evidence) and adds it to the underlying slice of data.
-func (ss *shareStack) peel(share []byte, delimited bool) (err error) {
-	if delimited {
-		var txLen uint64
-		share, txLen, err = ParseDelimiter(share)
-		if err != nil {
-			return err
-		}
-		if txLen == 0 {
-			return nil
-		}
-		ss.txLen = txLen
-	}
-	// safeLen describes the point in the share where it can be safely split. If
-	// split beyond this point, it is possible to break apart a length
-	// delimiter, which will result in incorrect share merging
-	safeLen := len(share) - binary.MaxVarintLen64
-	if safeLen < 0 {
-		safeLen = 0
-	}
-	if ss.txLen <= uint64(safeLen) {
-		ss.txs = append(ss.txs, share[:ss.txLen])
-		share = share[ss.txLen:]
-		return ss.peel(share, true)
-	}
-	// add the next share to the current share to continue merging if possible
-	if len(ss.shares) > ss.cursor+1 {
-		ss.cursor++
-		share := append(share, ss.shares[ss.cursor][consts.NamespaceSize+consts.ShareReservedBytes:]...)
-		return ss.peel(share, false)
-	}
-	// collect any remaining data
-	if ss.txLen <= uint64(len(share)) {
-		ss.txs = append(ss.txs, share[:ss.txLen])
-		share = share[ss.txLen:]
-		return ss.peel(share, true)
-	}
-	return errors.New("failure to parse block data: transaction length exceeded data length")
-}
-
-// parseMsgShares iterates through raw shares and separates the contiguous chunks
-// of data. It is only used for Messages, i.e. shares with a non-reserved namespace.
-func parseMsgShares(shares [][]byte) ([]coretypes.Message, error) {
-	if len(shares) == 0 {
-		return nil, nil
-	}
-
-	// set the first nid and current share
-	nid := shares[0][:consts.NamespaceSize]
-	currentShare := shares[0][consts.NamespaceSize:]
-	// find and remove the msg len delimiter
-	currentShare, msgLen, err := ParseDelimiter(currentShare)
-	if err != nil {
-		return nil, err
-	}
-
-	var msgs []coretypes.Message
-	for cursor := uint64(0); cursor < uint64(len(shares)); {
-		var msg coretypes.Message
-		currentShare, nid, cursor, msgLen, msg, err = nextMsg(
-			shares,
-			currentShare,
-			nid,
-			cursor,
-			msgLen,
-		)
-		if err != nil {
-			return nil, err
-		}
-		if msg.Data != nil {
-			msgs = append(msgs, msg)
-		}
-	}
-
-	return msgs, nil
-}
-
-func nextMsg(
-	shares [][]byte,
-	current,
-	nid []byte,
-	cursor,
-	msgLen uint64,
-) ([]byte, []byte, uint64, uint64, coretypes.Message, error) {
-	switch {
-	// the message uses all of the current share data and at least some of the
-	// next share
-	case msgLen > uint64(len(current)):
-		// add the next share to the current one and try again
-		cursor++
-		current = append(current, shares[cursor][consts.NamespaceSize:]...)
-		return nextMsg(shares, current, nid, cursor, msgLen)
-
-	// the msg we're looking for is contained in the current share
-	case msgLen <= uint64(len(current)):
-		msg := coretypes.Message{NamespaceID: nid, Data: current[:msgLen]}
-		cursor++
-
-		// call it a day if the work is done
-		if cursor >= uint64(len(shares)) {
-			return nil, nil, cursor, 0, msg, nil
-		}
-
-		nextNid := shares[cursor][:consts.NamespaceSize]
-		next, msgLen, err := ParseDelimiter(shares[cursor][consts.NamespaceSize:])
-		return next, nextNid, cursor, msgLen, msg, err
-	}
-	// this code is unreachable but the compiler doesn't know that
-	return nil, nil, 0, 0, coretypes.Message{}, nil
-}
-
-// ParseDelimiter finds and returns the length delimiter of the message provided
-// while also removing the delimiter bytes from the input
-func ParseDelimiter(input []byte) ([]byte, uint64, error) {
-	if len(input) == 0 {
-		return input, 0, nil
-	}
-
-	l := binary.MaxVarintLen64
-	if len(input) < binary.MaxVarintLen64 {
-		l = len(input)
-	}
-
-	delimiter := zeroPadIfNecessary(input[:l], binary.MaxVarintLen64)
-
-	// read the length of the message
-	r := bytes.NewBuffer(delimiter)
-	msgLen, err := binary.ReadUvarint(r)
-	if err != nil {
-		return nil, 0, err
-	}
-
-	// calculate the number of bytes used by the delimiter
-	lenBuf := make([]byte, binary.MaxVarintLen64)
-	n := binary.PutUvarint(lenBuf, msgLen)
-
-	// return the input without the length delimiter
-	return input[n:], msgLen, nil
-}
diff --git a/pkg/shares/share_splitting.go b/pkg/shares/share_splitting.go
index 4bff378db2..19f1fa621a 100644
--- a/pkg/shares/share_splitting.go
+++ b/pkg/shares/share_splitting.go
@@ -1,412 +1,120 @@
-package types
+package shares
 
 import (
-	"bytes"
 	"errors"
 	"fmt"
-	"math"
-	"sort"
 
-	"github.com/celestiaorg/nmt/namespace"
-	"github.com/tendermint/tendermint/libs/protoio"
 	"github.com/tendermint/tendermint/pkg/consts"
 	coretypes "github.com/tendermint/tendermint/types"
 )
 
-// MessageShareWriter lazily merges messages into shares that will eventually be
-// included in a data square. It also has methods to help progressively count
-// how many shares the messages written take up.
-type MessageShareWriter struct {
-	shares [][]NamespacedShare
-	count  int
-}
-
-func NewMessageShareWriter() *MessageShareWriter {
-	return &MessageShareWriter{}
-}
-
-// Write adds the delimited data to the underlying contiguous shares.
-func (msw *MessageShareWriter) Write(msg coretypes.Message) {
-	rawMsg, err := msg.MarshalDelimited()
-	if err != nil {
-		panic(fmt.Sprintf("app accepted a Message that can not be encoded %#v", msg))
-	}
-	newShares := make([]NamespacedShare, 0)
-	newShares = AppendToShares(newShares, msg.NamespaceID, rawMsg)
-	msw.shares = append(msw.shares, newShares)
-	msw.count += len(newShares)
-}
-
-// Export finalizes and returns the underlying contiguous shares.
-func (msw *MessageShareWriter) Export() NamespacedShares {
-	msw.sortMsgs()
-	shares := make([]NamespacedShare, msw.count)
-	cursor := 0
-	for _, messageShares := range msw.shares {
-		for _, share := range messageShares {
-			shares[cursor] = share
-			cursor++
-		}
-	}
-	return shares
-}
-
-func (msw *MessageShareWriter) sortMsgs() {
-	sort.Slice(msw.shares, func(i, j int) bool {
-		return bytes.Compare(msw.shares[i][0].ID, msw.shares[j][0].ID) < 0
-	})
-}
-
-// Count returns the current number of shares that will be made if exporting.
-func (msw *MessageShareWriter) Count() int {
-	return msw.count
-}
-
-// appendToShares appends raw data as shares.
-// Used for messages.
-func AppendToShares(shares []NamespacedShare, nid namespace.ID, rawData []byte) []NamespacedShare {
-	if len(rawData) <= consts.MsgShareSize {
-		rawShare := append(append(
-			make([]byte, 0, len(nid)+len(rawData)),
-			nid...),
-			rawData...,
-		)
-		paddedShare := zeroPadIfNecessary(rawShare, consts.ShareSize)
-		share := NamespacedShare{paddedShare, nid}
-		shares = append(shares, share)
-	} else { // len(rawData) > MsgShareSize
-		shares = append(shares, splitMessage(rawData, nid)...)
-	}
-	return shares
-}
-
-// splitMessage breaks the data in a message into the minimum number of
-// namespaced shares
-func splitMessage(rawData []byte, nid namespace.ID) NamespacedShares {
-	shares := make([]NamespacedShare, 0)
-	firstRawShare := append(append(
-		make([]byte, 0, consts.ShareSize),
-		nid...),
-		rawData[:consts.MsgShareSize]...,
+var (
+	ErrIncorrectNumberOfIndexes = errors.New(
+		"number of malleated transactions is not identical to the number of wrapped transactions",
 	)
-	shares = append(shares, NamespacedShare{firstRawShare, nid})
-	rawData = rawData[consts.MsgShareSize:]
-	for len(rawData) > 0 {
-		shareSizeOrLen := min(consts.MsgShareSize, len(rawData))
-		rawShare := append(append(
-			make([]byte, 0, consts.ShareSize),
-			nid...),
-			rawData[:shareSizeOrLen]...,
-		)
-		paddedShare := zeroPadIfNecessary(rawShare, consts.ShareSize)
-		share := NamespacedShare{paddedShare, nid}
-		shares = append(shares, share)
-		rawData = rawData[shareSizeOrLen:]
-	}
-	return shares
-}
-
-// ContiguousShareWriter will write raw data contiguously across a progressively
-// increasing set of shares. It is used to lazily split block data such as transactions
-// into shares.
-type ContiguousShareWriter struct {
-	shares       []NamespacedShare
-	pendingShare NamespacedShare
-	namespace    namespace.ID
-}
-
-// NewContiguousShareWriter returns a ContigousShareWriter using the provided
-// namespace.
-func NewContiguousShareWriter(ns namespace.ID) *ContiguousShareWriter {
-	pendingShare := NamespacedShare{ID: ns, Share: make([]byte, 0, consts.ShareSize)}
-	pendingShare.Share = append(pendingShare.Share, ns...)
-	return &ContiguousShareWriter{pendingShare: pendingShare, namespace: ns}
-}
-
-// Write adds the delimited data to the underlying contiguous shares.
-func (csw *ContiguousShareWriter) Write(rawData []byte) {
-	// if this is the first time writing to a pending share, we must add the
-	// reserved bytes
-	if len(csw.pendingShare.Share) == consts.NamespaceSize {
-		csw.pendingShare.Share = append(csw.pendingShare.Share, 0)
-	}
-
-	txCursor := len(rawData)
-	for txCursor != 0 {
-		// find the len left in the pending share
-		pendingLeft := consts.ShareSize - len(csw.pendingShare.Share)
-
-		// if we can simply add the tx to the share without creating a new
-		// pending share, do so and return
-		if len(rawData) <= pendingLeft {
-			csw.pendingShare.Share = append(csw.pendingShare.Share, rawData...)
-			break
-		}
-
-		// if we can only add a portion of the transaction to the pending share,
-		// then we add it and add the pending share to the finalized shares.
-		chunk := rawData[:pendingLeft]
-		csw.pendingShare.Share = append(csw.pendingShare.Share, chunk...)
-		csw.stackPending()
-
-		// update the cursor
-		rawData = rawData[pendingLeft:]
-		txCursor = len(rawData)
-
-		// add the share reserved bytes to the new pending share
-		pendingCursor := len(rawData) + consts.NamespaceSize + consts.ShareReservedBytes
-		var reservedByte byte
-		if pendingCursor >= consts.ShareSize {
-			// the share reserve byte is zero when some contiguously written
-			// data takes up the entire share
-			reservedByte = byte(0)
-		} else {
-			reservedByte = byte(pendingCursor)
-		}
+	ErrUnexpectedFirstMessageShareIndex = errors.New(
+		"the first message started at an unexpected index",
+	)
+)
 
-		csw.pendingShare.Share = append(csw.pendingShare.Share, reservedByte)
+func Split(data coretypes.Data) ([][]byte, error) {
+	if data.OriginalSquareSize == 0 || !isPowerOf2(data.OriginalSquareSize) {
+		return nil, fmt.Errorf("square size is not a power of two: %d", data.OriginalSquareSize)
 	}
+	wantShareCount := int(data.OriginalSquareSize * data.OriginalSquareSize)
+	currentShareCount := 0
 
-	// if the share is exactly the correct size, then append to shares
-	if len(csw.pendingShare.Share) == consts.ShareSize {
-		csw.stackPending()
-	}
-}
+	txShares := SplitTxs(data.Txs)
+	currentShareCount += len(txShares)
 
-// stackPending will add the pending share to accumlated shares provided that it is long enough
-func (csw *ContiguousShareWriter) stackPending() {
-	if len(csw.pendingShare.Share) < consts.ShareSize {
-		return
-	}
-	csw.shares = append(csw.shares, csw.pendingShare)
-	newPendingShare := make([]byte, 0, consts.ShareSize)
-	newPendingShare = append(newPendingShare, csw.namespace...)
-	csw.pendingShare = NamespacedShare{
-		Share: newPendingShare,
-		ID:    csw.namespace,
+	evdShares, err := SplitEvidence(data.Evidence.Evidence)
+	if err != nil {
+		return nil, err
 	}
-}
+	currentShareCount += len(evdShares)
 
-// Export finalizes and returns the underlying contiguous shares.
-func (csw *ContiguousShareWriter) Export() NamespacedShares {
-	// add the pending share to the current shares before returning
-	if len(csw.pendingShare.Share) > consts.NamespaceSize {
-		csw.pendingShare.Share = zeroPadIfNecessary(csw.pendingShare.Share, consts.ShareSize)
-		csw.shares = append(csw.shares, csw.pendingShare)
-	}
-	// force the last share to have a reserve byte of zero
-	if len(csw.shares) == 0 {
-		return csw.shares
-	}
-	lastShare := csw.shares[len(csw.shares)-1]
-	rawLastShare := lastShare.Data()
+	// msgIndexes will be nil if we are working with a list of txs that do not
+	// have a msg index. this preserves backwards compatibility with old blocks
+	// that do not follow the non-interactive defaults
+	msgIndexes := ExtractShareIndexes(data.Txs)
 
-	for i := 0; i < consts.ShareReservedBytes; i++ {
-		// here we force the last share reserved byte to be zero to avoid any
-		// confusion for light clients parsing these shares, as the rest of the
-		// data after transaction is padding. See
-		// https://github.com/celestiaorg/celestia-specs/blob/master/src/specs/data_structures.md#share
-		rawLastShare[consts.NamespaceSize+i] = byte(0)
+	var msgShares [][]byte
+	if msgIndexes != nil && int(msgIndexes[0]) != currentShareCount {
+		return nil, ErrUnexpectedFirstMessageShareIndex
 	}
 
-	newLastShare := NamespacedShare{
-		Share: rawLastShare,
-		ID:    lastShare.NamespaceID(),
+	msgShares, err = SplitMessages(msgIndexes, data.Messages.MessagesList)
+	if err != nil {
+		return nil, err
 	}
-	csw.shares[len(csw.shares)-1] = newLastShare
-	return csw.shares
-}
+	currentShareCount += len(msgShares)
 
-// Count returns the current number of shares that will be made if exporting.
-func (csw *ContiguousShareWriter) Count() (count, availableBytes int) {
-	availableBytes = consts.TxShareSize - (len(csw.pendingShare.Share) - consts.NamespaceSize)
-	return len(csw.shares), availableBytes
-}
+	tailShares := TailPaddingShares(wantShareCount - currentShareCount).RawShares()
 
-// tail is filler for all tail padded shares
-// it is allocated once and used everywhere
-var tailPaddingShare = append(
-	append(make([]byte, 0, consts.ShareSize), consts.TailPaddingNamespaceID...),
-	bytes.Repeat([]byte{0}, consts.ShareSize-consts.NamespaceSize)...,
-)
+	// todo: optimize using a predefined slice
+	shares := append(append(append(
+		txShares,
+		evdShares...),
+		msgShares...),
+		tailShares...)
 
-func TailPaddingShares(n int) NamespacedShares {
-	shares := make([]NamespacedShare, n)
-	for i := 0; i < n; i++ {
-		shares[i] = NamespacedShare{
-			Share: tailPaddingShare,
-			ID:    consts.TailPaddingNamespaceID,
+	return shares, nil
+}
+
+// ExtractShareIndexes iterates over the transactions and extracts the share
+// indexes from wrapped transactions. It returns nil if the transactions are
+// from an old block that did not have share indexes in the wrapped txs.
+func ExtractShareIndexes(txs coretypes.Txs) []uint32 {
+	var msgIndexes []uint32
+	for _, rawTx := range txs {
+		if malleatedTx, isMalleated := coretypes.UnwrapMalleatedTx(rawTx); isMalleated {
+			// Since share index == 0 is invalid, it indicates that we are
+			// attempting to extract share indexes from txs that do not have any
+			// due to them being old. here we return nil to indicate that we are
+			// attempting to extract indexes from a block that doesn't support
+			// it. It's check for 0 because if there is a message in the block,
+			// then there must also be a tx, which will take up at least one
+			// share.
+			if malleatedTx.ShareIndex == 0 {
+				return nil
+			}
+			msgIndexes = append(msgIndexes, malleatedTx.ShareIndex)
 		}
 	}
-	return shares
-}
 
-func min(a, b int) int {
-	if a <= b {
-		return a
-	}
-	return b
+	return msgIndexes
 }
 
-func zeroPadIfNecessary(share []byte, width int) []byte {
-	oldLen := len(share)
-	if oldLen < width {
-		missingBytes := width - oldLen
-		padByte := []byte{0}
-		padding := bytes.Repeat(padByte, missingBytes)
-		share = append(share, padding...)
-		return share
+func SplitTxs(txs coretypes.Txs) [][]byte {
+	writer := NewContiguousShareSplitter(consts.TxNamespaceID)
+	for _, tx := range txs {
+		writer.WriteTx(tx)
 	}
-	return share
+	return writer.Export().RawShares()
 }
 
-func SplitTxsIntoShares(txs coretypes.Txs) NamespacedShares {
-	rawDatas := make([][]byte, len(txs))
-	for i, tx := range txs {
-		rawData, err := tx.MarshalDelimited()
+func SplitEvidence(evd coretypes.EvidenceList) ([][]byte, error) {
+	writer := NewContiguousShareSplitter(consts.EvidenceNamespaceID)
+	var err error
+	for _, ev := range evd {
+		err = writer.WriteEvidence(ev)
 		if err != nil {
-			panic(fmt.Sprintf("included Tx in mem-pool that can not be encoded %v", tx))
+			return nil, err
 		}
-		rawDatas[i] = rawData
 	}
-
-	w := NewContiguousShareWriter(consts.TxNamespaceID)
-	for _, tx := range rawDatas {
-		w.Write(tx)
-	}
-
-	return w.Export()
+	return writer.Export().RawShares(), nil
 }
 
-func SplitEvidenceIntoShares(data *coretypes.EvidenceData) NamespacedShares {
-	rawDatas := make([][]byte, 0, len(data.Evidence))
-	for _, ev := range data.Evidence {
-		pev, err := coretypes.EvidenceToProto(ev)
-		if err != nil {
-			panic("failure to convert evidence to equivalent proto type")
-		}
-		rawData, err := protoio.MarshalDelimited(pev)
-		if err != nil {
-			panic(err)
-		}
-		rawDatas = append(rawDatas, rawData)
+func SplitMessages(indexes []uint32, msgs []coretypes.Message) ([][]byte, error) {
+	if indexes != nil && len(indexes) != len(msgs) {
+		return nil, ErrIncorrectNumberOfIndexes
 	}
-	w := NewContiguousShareWriter(consts.EvidenceNamespaceID)
-	for _, evd := range rawDatas {
-		w.Write(evd)
-	}
-	return w.Export()
-}
-
-func SplitMessagesIntoShares(msgs coretypes.Messages) NamespacedShares {
-	shares := make([]NamespacedShare, 0)
-	msgs.SortMessages()
-	for _, m := range msgs.MessagesList {
-		rawData, err := m.MarshalDelimited()
-		if err != nil {
-			panic(fmt.Sprintf("app accepted a Message that can not be encoded %#v", m))
-		}
-		shares = AppendToShares(shares, m.NamespaceID, rawData)
-	}
-	return shares
-}
-
-// SortMessages sorts messages by ascending namespace id
-func SortMessages(msgs *coretypes.Messages) {
-	sort.SliceStable(msgs.MessagesList, func(i, j int) bool {
-		return bytes.Compare(msgs.MessagesList[i].NamespaceID, msgs.MessagesList[j].NamespaceID) < 0
-	})
-}
-
-// ComputeShares splits block data into shares of an original data square and
-// returns them along with an amount of non-redundant shares. If a square size
-// of 0 is passed, then it is determined based on how many shares are needed to
-// fill the square for the underlying block data. The square size is stored in
-// the local instance of the struct.
-func ComputeShares(data *coretypes.Data, squareSize uint64) (NamespacedShares, int, error) {
-	if squareSize != 0 {
-		if !powerOf2(squareSize) {
-			return nil, 0, errors.New("square size is not a power of two")
+	writer := NewMessageShareSplitter()
+	for i, msg := range msgs {
+		writer.Write(msg)
+		if indexes != nil && len(indexes) > i+1 {
+			writer.WriteNamespacedPaddedShares(int(indexes[i+1]) - writer.Count())
 		}
 	}
-
-	// reserved shares:
-	txShares := SplitTxsIntoShares(data.Txs)
-	evidenceShares := SplitEvidenceIntoShares(&data.Evidence)
-
-	// application data shares from messages:
-	msgShares := SplitMessagesIntoShares(data.Messages)
-	curLen := len(txShares) + len(evidenceShares) + len(msgShares)
-
-	if curLen > consts.MaxShareCount {
-		panic(fmt.Sprintf("Block data exceeds the max square size. Number of shares required: %d\n", curLen))
-	}
-
-	// find the number of shares needed to create a square that has a power of
-	// two width
-	wantLen := int(squareSize * squareSize)
-	if squareSize == 0 {
-		wantLen = paddedLen(curLen)
-	}
-
-	if wantLen < curLen {
-		return nil, 0, errors.New("square size too small to fit block data")
-	}
-
-	// ensure that the min square size is used
-	if wantLen < consts.MinSharecount {
-		wantLen = consts.MinSharecount
-	}
-
-	tailShares := TailPaddingShares(wantLen - curLen)
-
-	shares := append(append(append(
-		txShares,
-		evidenceShares...),
-		msgShares...),
-		tailShares...)
-
-	if squareSize == 0 {
-		squareSize = uint64(math.Sqrt(float64(wantLen)))
-	}
-
-	data.OriginalSquareSize = squareSize
-
-	return shares, curLen, nil
-}
-
-// paddedLen calculates the number of shares needed to make a power of 2 square
-// given the current number of shares
-func paddedLen(length int) int {
-	width := uint32(math.Ceil(math.Sqrt(float64(length))))
-	width = nextHighestPowerOf2(width)
-	return int(width * width)
-}
-
-// nextPowerOf2 returns the next highest power of 2 unless the input is a power
-// of two, in which case it returns the input
-func nextHighestPowerOf2(v uint32) uint32 {
-	if v == 0 {
-		return 0
-	}
-
-	// find the next highest power using bit mashing
-	v--
-	v |= v >> 1
-	v |= v >> 2
-	v |= v >> 4
-	v |= v >> 8
-	v |= v >> 16
-	v++
-
-	// return the next highest power
-	return v
-}
-
-// powerOf2 checks if number is power of 2
-func powerOf2(v uint64) bool {
-	if v&(v-1) == 0 && v != 0 {
-		return true
-	}
-	return false
+	return writer.Export().RawShares(), nil
 }
diff --git a/pkg/shares/shares.go b/pkg/shares/shares.go
index 077d35db6d..55991641a6 100644
--- a/pkg/shares/shares.go
+++ b/pkg/shares/shares.go
@@ -1,4 +1,4 @@
-package types
+package shares
 
 import (
 	"encoding/binary"
diff --git a/pkg/shares/shares_test.go b/pkg/shares/shares_test.go
index 62cd65c7f2..be08f215d9 100644
--- a/pkg/shares/shares_test.go
+++ b/pkg/shares/shares_test.go
@@ -1,47 +1,44 @@
-package types
+package shares
 
 import (
-	"bytes"
 	"context"
-	"fmt"
 	"math"
 	"math/rand"
 	"reflect"
-	"sort"
 	"testing"
 	"time"
 
 	"github.com/celestiaorg/rsmt2d"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	tmrand "github.com/tendermint/tendermint/libs/rand"
 	"github.com/tendermint/tendermint/pkg/consts"
 	coretypes "github.com/tendermint/tendermint/types"
 )
 
-type Splitter interface {
-	SplitIntoShares() NamespacedShares
-}
+// var defaultVoteTime = time.Date(2019, 1, 1, 0, 0, 0, 0, time.UTC)
 
+// TODO: refactor into different tests
 // func TestMakeShares(t *testing.T) {
 // 	reservedTxNamespaceID := append(bytes.Repeat([]byte{0}, 7), 1)
 // 	reservedEvidenceNamespaceID := append(bytes.Repeat([]byte{0}, 7), 3)
-// 	val := NewMockPV()
+// 	val := coretypes.NewMockPV()
 // 	blockID := makeBlockID([]byte("blockhash"), 1000, []byte("partshash"))
 // 	blockID2 := makeBlockID([]byte("blockhash2"), 1000, []byte("partshash"))
 // 	vote1 := makeVote(t, val, "chainID", 0, 10, 2, 1, blockID, defaultVoteTime)
 // 	vote2 := makeVote(t, val, "chainID", 0, 10, 2, 1, blockID2, defaultVoteTime)
-// 	testEvidence := &DuplicateVoteEvidence{
+// 	testEvidence := &coretypes.DuplicateVoteEvidence{
 // 		VoteA: vote1,
 // 		VoteB: vote2,
 // 	}
-// 	protoTestEvidence, err := EvidenceToProto(testEvidence)
+// 	protoTestEvidence, err := coretypes.EvidenceToProto(testEvidence)
 // 	if err != nil {
 // 		t.Error(err)
 // 	}
 // 	testEvidenceBytes, err := protoio.MarshalDelimited(protoTestEvidence)
-// 	largeTx := Tx(bytes.Repeat([]byte("large Tx"), 50))
+// 	largeTx := coretypes.Tx(bytes.Repeat([]byte("large Tx"), 50))
 // 	largeTxLenDelimited, _ := largeTx.MarshalDelimited()
-// 	smolTx := Tx("small Tx")
+// 	smolTx := coretypes.Tx("small Tx")
 // 	smolTxLenDelimited, _ := smolTx.MarshalDelimited()
 // 	msg1 := coretypes.Message{
 // 		NamespaceID: namespace.ID("8bytesss"),
@@ -64,7 +61,7 @@ type Splitter interface {
 // 			name: "evidence",
 // 			args: args{
 // 				data: &coretypes.EvidenceData{
-// 					Evidence: []Evidence{testEvidence},
+// 					Evidence: []coretypes.Evidence{testEvidence},
 // 				},
 // 			},
 // 			want: NamespacedShares{
@@ -86,7 +83,7 @@ type Splitter interface {
 // 		},
 // 		{"small LL Tx",
 // 			args{
-// 				data: Txs{smolTx},
+// 				data: coretypes.Txs{smolTx},
 // 			},
 // 			NamespacedShares{
 // 				NamespacedShare{
@@ -100,7 +97,7 @@ type Splitter interface {
 // 		},
 // 		{"one large LL Tx",
 // 			args{
-// 				data: Txs{largeTx},
+// 				data: coretypes.Txs{largeTx},
 // 			},
 // 			NamespacedShares{
 // 				NamespacedShare{
@@ -121,7 +118,7 @@ type Splitter interface {
 // 		},
 // 		{"large then small LL Tx",
 // 			args{
-// 				data: Txs{largeTx, smolTx},
+// 				data: coretypes.Txs{largeTx, smolTx},
 // 			},
 // 			NamespacedShares{
 // 				NamespacedShare{
@@ -148,7 +145,7 @@ type Splitter interface {
 // 		},
 // 		{"ll-app message",
 // 			args{
-// 				data: Messages{[]coretypes.coretypes.Message{msg1}},
+// 				data: coretypes.Messages{[]coretypes.Message{msg1}},
 // 			},
 // 			NamespacedShares{
 // 				NamespacedShare{
@@ -197,24 +194,7 @@ func Test_zeroPadIfNecessary(t *testing.T) {
 	}
 }
 
-func Test_appendToSharesOverwrite(t *testing.T) {
-	var shares NamespacedShares
-
-	// generate some arbitrary namespaced shares first share that must be split
-	newShare := generateRandomNamespacedShares(1, consts.MsgShareSize+1)[0]
-
-	// make a copy of the portion of the share to check if it's overwritten later
-	extraCopy := make([]byte, consts.MsgShareSize)
-	copy(extraCopy, newShare.Share[:consts.MsgShareSize])
-
-	// use appendToShares to add our new share
-	AppendToShares(shares, newShare.ID, newShare.Share)
-
-	// check if the original share data has been overwritten.
-	assert.Equal(t, extraCopy, []byte(newShare.Share[:consts.MsgShareSize]))
-}
-
-func TestDataFromSquare(t *testing.T) {
+func TestMerge(t *testing.T) {
 	type test struct {
 		name     string
 		txCount  int
@@ -245,7 +225,7 @@ func TestDataFromSquare(t *testing.T) {
 				tc.maxSize,
 			)
 
-			shares, _, err := ComputeShares(&data, 0)
+			shares, _, err := data.ComputeShares(0)
 			require.NoError(t, err)
 			rawShares := shares.RawShares()
 
@@ -254,7 +234,7 @@ func TestDataFromSquare(t *testing.T) {
 				t.Error(err)
 			}
 
-			res, err := DataFromSquare(eds)
+			res, err := Merge(eds)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -279,84 +259,7 @@ func TestDataFromSquare(t *testing.T) {
 	}
 }
 
-func TestFuzz_DataFromSquare(t *testing.T) {
-	t.Skip()
-	// run random shares through processContiguousShares for a minute
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
-	defer cancel()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			TestDataFromSquare(t)
-		}
-	}
-}
-
-func Test_processContiguousShares(t *testing.T) {
-	// exactTxShareSize is the length of tx that will fit exactly into a single
-	// share, accounting for namespace id and the length delimiter prepended to
-	// each tx
-	const exactTxShareSize = consts.TxShareSize - 1
-
-	type test struct {
-		name    string
-		txSize  int
-		txCount int
-	}
-
-	// each test is ran twice, once using txSize as an exact size, and again
-	// using it as a cap for randomly sized txs
-	tests := []test{
-		{"single small tx", 10, 1},
-		{"many small txs", 10, 10},
-		{"single big tx", 1000, 1},
-		{"many big txs", 1000, 10},
-		{"single exact size tx", exactTxShareSize, 1},
-		{"many exact size txs", exactTxShareSize, 10},
-	}
-
-	for _, tc := range tests {
-		tc := tc
-
-		// run the tests with identically sized txs
-		t.Run(fmt.Sprintf("%s idendically sized ", tc.name), func(t *testing.T) {
-			txs := generateRandomContiguousShares(tc.txCount, tc.txSize)
-
-			shares := txs.SplitIntoShares()
-
-			parsedTxs, err := processContiguousShares(shares.RawShares())
-			if err != nil {
-				t.Error(err)
-			}
-
-			// check that the data parsed is identical
-			for i := 0; i < len(txs); i++ {
-				assert.Equal(t, []byte(txs[i]), parsedTxs[i])
-			}
-		})
-
-		// run the same tests using randomly sized txs with caps of tc.txSize
-		t.Run(fmt.Sprintf("%s randomly sized", tc.name), func(t *testing.T) {
-			txs := generateRandomlySizedContiguousShares(tc.txCount, tc.txSize)
-
-			shares := txs.SplitIntoShares()
-
-			parsedTxs, err := processContiguousShares(shares.RawShares())
-			if err != nil {
-				t.Error(err)
-			}
-
-			// check that the data parsed is identical to the original
-			for i := 0; i < len(txs); i++ {
-				assert.Equal(t, []byte(txs[i]), parsedTxs[i])
-			}
-		})
-	}
-}
-
-func TestFuzz_processContiguousShares(t *testing.T) {
+func TestFuzz_Merge(t *testing.T) {
 	t.Skip()
 	// run random shares through processContiguousShares for a minute
 	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
@@ -366,108 +269,8 @@ func TestFuzz_processContiguousShares(t *testing.T) {
 		case <-ctx.Done():
 			return
 		default:
-			Test_processContiguousShares(t)
-		}
-	}
-}
-
-func Test_parseMsgShares(t *testing.T) {
-	// exactMsgShareSize is the length of message that will fit exactly into a single
-	// share, accounting for namespace id and the length delimiter prepended to
-	// each message
-	const exactMsgShareSize = consts.MsgShareSize - 2
-
-	type test struct {
-		name     string
-		msgSize  int
-		msgCount int
-	}
-
-	// each test is ran twice, once using msgSize as an exact size, and again
-	// using it as a cap for randomly sized leaves
-	tests := []test{
-		{"single small msg", 1, 1},
-		{"many small msgs", 4, 10},
-		{"single big msg", 1000, 1},
-		{"many big msgs", 1000, 10},
-		{"single exact size msg", exactMsgShareSize, 1},
-		{"many exact size msgs", exactMsgShareSize, 10},
-	}
-
-	for _, tc := range tests {
-		tc := tc
-
-		// run the tests with identically sized messagses
-		t.Run(fmt.Sprintf("%s idendically sized ", tc.name), func(t *testing.T) {
-			rawmsgs := make([]coretypes.Message, tc.msgCount)
-			for i := 0; i < tc.msgCount; i++ {
-				rawmsgs[i] = generateRandomMessage(tc.msgSize)
-			}
-			msgs := coretypes.Messages{MessagesList: rawmsgs}
-
-			shares := msgs.SplitIntoShares()
-
-			parsedMsgs, err := parseMsgShares(shares.RawShares())
-			if err != nil {
-				t.Error(err)
-			}
-
-			// check that the namesapces and data are the same
-			for i := 0; i < len(msgs.MessagesList); i++ {
-				assert.Equal(t, msgs.MessagesList[i].NamespaceID, parsedMsgs[i].NamespaceID)
-				assert.Equal(t, msgs.MessagesList[i].Data, parsedMsgs[i].Data)
-			}
-		})
-
-		// run the same tests using randomly sized messages with caps of tc.msgSize
-		t.Run(fmt.Sprintf("%s randomly sized", tc.name), func(t *testing.T) {
-			msgs := generateRandomlySizedMessages(tc.msgCount, tc.msgSize)
-			shares := msgs.SplitIntoShares()
-
-			parsedMsgs, err := parseMsgShares(shares.RawShares())
-			if err != nil {
-				t.Error(err)
-			}
-
-			// check that the namesapces and data are the same
-			for i := 0; i < len(msgs.MessagesList); i++ {
-				assert.Equal(t, msgs.MessagesList[i].NamespaceID, parsedMsgs[i].NamespaceID)
-				assert.Equal(t, msgs.MessagesList[i].Data, parsedMsgs[i].Data)
-			}
-		})
-	}
-}
-
-func TestContigShareWriter(t *testing.T) {
-	// note that this test is mainly for debugging purposes, the main round trip
-	// tests occur in TestDataFromSquare and Test_processContiguousShares
-	w := NewContiguousShareWriter(consts.TxNamespaceID)
-	txs := generateRandomContiguousShares(33, 200)
-	for _, tx := range txs {
-		rawTx, _ := tx.MarshalDelimited()
-		w.Write(rawTx)
-	}
-	resShares := w.Export()
-	rawResTxs, err := processContiguousShares(resShares.RawShares())
-	resTxs := coretypes.ToTxs(rawResTxs)
-	require.NoError(t, err)
-
-	assert.Equal(t, txs, resTxs)
-}
-
-func Test_parseDelimiter(t *testing.T) {
-	for i := uint64(0); i < 100; i++ {
-		tx := generateRandomContiguousShares(1, int(i))[0]
-		input, err := tx.MarshalDelimited()
-		if err != nil {
-			panic(err)
-		}
-		res, txLen, err := ParseDelimiter(input)
-		if err != nil {
-			panic(err)
+			TestMerge(t)
 		}
-		assert.Equal(t, i, txLen)
-		assert.Equal(t, []byte(tx), res)
 	}
 }
 
@@ -518,6 +321,9 @@ func generateRandomlySizedMessages(count, maxMsgSize int) coretypes.Messages {
 	msgs := make([]coretypes.Message, count)
 	for i := 0; i < count; i++ {
 		msgs[i] = generateRandomMessage(rand.Intn(maxMsgSize))
+		if len(msgs[i].Data) == 0 {
+			i--
+		}
 	}
 
 	// this is just to let us use assert.Equal
@@ -531,43 +337,9 @@ func generateRandomlySizedMessages(count, maxMsgSize int) coretypes.Messages {
 }
 
 func generateRandomMessage(size int) coretypes.Message {
-	share := generateRandomNamespacedShares(1, size)[0]
 	msg := coretypes.Message{
-		NamespaceID: share.NamespaceID(),
-		Data:        share.Data(),
+		NamespaceID: tmrand.Bytes(consts.NamespaceSize),
+		Data:        tmrand.Bytes(size),
 	}
 	return msg
 }
-
-func generateRandomNamespacedShares(count, msgSize int) NamespacedShares {
-	shares := generateRandNamespacedRawData(uint32(count), consts.NamespaceSize, uint32(msgSize))
-	msgs := make([]coretypes.Message, count)
-	for i, s := range shares {
-		msgs[i] = coretypes.Message{
-			Data:        s[consts.NamespaceSize:],
-			NamespaceID: s[:consts.NamespaceSize],
-		}
-	}
-	return SplitMessagesIntoShares(coretypes.Messages{MessagesList: msgs})
-}
-
-func generateRandNamespacedRawData(total, nidSize, leafSize uint32) [][]byte {
-	data := make([][]byte, total)
-	for i := uint32(0); i < total; i++ {
-		nid := make([]byte, nidSize)
-		rand.Read(nid)
-		data[i] = nid
-	}
-	sortByteArrays(data)
-	for i := uint32(0); i < total; i++ {
-		d := make([]byte, leafSize)
-		rand.Read(d)
-		data[i] = append(data[i], d...)
-	}
-
-	return data
-}
-
-func sortByteArrays(src [][]byte) {
-	sort.Slice(src, func(i, j int) bool { return bytes.Compare(src[i], src[j]) < 0 })
-}
diff --git a/pkg/shares/split_contiguous_shares.go b/pkg/shares/split_contiguous_shares.go
new file mode 100644
index 0000000000..ff3fc32ab0
--- /dev/null
+++ b/pkg/shares/split_contiguous_shares.go
@@ -0,0 +1,190 @@
+package shares
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/celestiaorg/nmt/namespace"
+	"github.com/tendermint/tendermint/libs/protoio"
+	"github.com/tendermint/tendermint/pkg/consts"
+	coretypes "github.com/tendermint/tendermint/types"
+)
+
+// ContiguousShareSplitter will write raw data contiguously across a progressively
+// increasing set of shares. It is used to lazily split block data such as transactions
+// into shares.
+type ContiguousShareSplitter struct {
+	shares       []NamespacedShare
+	pendingShare NamespacedShare
+	namespace    namespace.ID
+}
+
+// NewContiguousShareSplitter returns a ContiguousShareSplitter using the provided
+// namespace.
+func NewContiguousShareSplitter(ns namespace.ID) *ContiguousShareSplitter {
+	pendingShare := NamespacedShare{ID: ns, Share: make([]byte, 0, consts.ShareSize)}
+	pendingShare.Share = append(pendingShare.Share, ns...)
+	return &ContiguousShareSplitter{pendingShare: pendingShare, namespace: ns}
+}
+
+func (csw *ContiguousShareSplitter) WriteTx(tx coretypes.Tx) {
+	rawData, err := tx.MarshalDelimited()
+	if err != nil {
+		panic(fmt.Sprintf("included Tx in mem-pool that can not be encoded %v", tx))
+	}
+	csw.WriteBytes(rawData)
+}
+
+func (csw *ContiguousShareSplitter) WriteEvidence(evd coretypes.Evidence) error {
+	pev, err := coretypes.EvidenceToProto(evd)
+	if err != nil {
+		return err
+	}
+	rawData, err := protoio.MarshalDelimited(pev)
+	if err != nil {
+		return err
+	}
+	csw.WriteBytes(rawData)
+	return nil
+}
+
+// WriteBytes adds the delimited data to the underlying contiguous shares.
+func (csw *ContiguousShareSplitter) WriteBytes(rawData []byte) {
+	// if this is the first time writing to a pending share, we must add the
+	// reserved bytes
+	if len(csw.pendingShare.Share) == consts.NamespaceSize {
+		csw.pendingShare.Share = append(csw.pendingShare.Share, 0)
+	}
+
+	txCursor := len(rawData)
+	for txCursor != 0 {
+		// find the len left in the pending share
+		pendingLeft := consts.ShareSize - len(csw.pendingShare.Share)
+
+		// if we can simply add the tx to the share without creating a new
+		// pending share, do so and return
+		if len(rawData) <= pendingLeft {
+			csw.pendingShare.Share = append(csw.pendingShare.Share, rawData...)
+			break
+		}
+
+		// if we can only add a portion of the transaction to the pending share,
+		// then we add it and add the pending share to the finalized shares.
+		chunk := rawData[:pendingLeft]
+		csw.pendingShare.Share = append(csw.pendingShare.Share, chunk...)
+		csw.stackPending()
+
+		// update the cursor
+		rawData = rawData[pendingLeft:]
+		txCursor = len(rawData)
+
+		// add the share reserved bytes to the new pending share
+		pendingCursor := len(rawData) + consts.NamespaceSize + consts.ShareReservedBytes
+		var reservedByte byte
+		if pendingCursor >= consts.ShareSize {
+			// the share reserve byte is zero when some contiguously written
+			// data takes up the entire share
+			reservedByte = byte(0)
+		} else {
+			reservedByte = byte(pendingCursor)
+		}
+
+		csw.pendingShare.Share = append(csw.pendingShare.Share, reservedByte)
+	}
+
+	// if the share is exactly the correct size, then append to shares
+	if len(csw.pendingShare.Share) == consts.ShareSize {
+		csw.stackPending()
+	}
+}
+
+// stackPending will add the pending share to accumlated shares provided that it is long enough
+func (csw *ContiguousShareSplitter) stackPending() {
+	if len(csw.pendingShare.Share) < consts.ShareSize {
+		return
+	}
+	csw.shares = append(csw.shares, csw.pendingShare)
+	newPendingShare := make([]byte, 0, consts.ShareSize)
+	newPendingShare = append(newPendingShare, csw.namespace...)
+	csw.pendingShare = NamespacedShare{
+		Share: newPendingShare,
+		ID:    csw.namespace,
+	}
+}
+
+// Export finalizes and returns the underlying contiguous shares.
+func (csw *ContiguousShareSplitter) Export() NamespacedShares {
+	// add the pending share to the current shares before returning
+	if len(csw.pendingShare.Share) > consts.NamespaceSize {
+		csw.pendingShare.Share = zeroPadIfNecessary(csw.pendingShare.Share, consts.ShareSize)
+		csw.shares = append(csw.shares, csw.pendingShare)
+	}
+	// force the last share to have a reserve byte of zero
+	if len(csw.shares) == 0 {
+		return csw.shares
+	}
+	lastShare := csw.shares[len(csw.shares)-1]
+	rawLastShare := lastShare.Data()
+
+	for i := 0; i < consts.ShareReservedBytes; i++ {
+		// here we force the last share reserved byte to be zero to avoid any
+		// confusion for light clients parsing these shares, as the rest of the
+		// data after transaction is padding. See
+		// https://github.com/celestiaorg/celestia-specs/blob/master/src/specs/data_structures.md#share
+		rawLastShare[consts.NamespaceSize+i] = byte(0)
+	}
+
+	newLastShare := NamespacedShare{
+		Share: rawLastShare,
+		ID:    lastShare.NamespaceID(),
+	}
+	csw.shares[len(csw.shares)-1] = newLastShare
+	return csw.shares
+}
+
+// Count returns the current number of shares that will be made if exporting.
+func (csw *ContiguousShareSplitter) Count() (count, availableBytes int) {
+	if len(csw.pendingShare.Share) > consts.NamespaceSize {
+		return len(csw.shares), 0
+	}
+	availableBytes = consts.TxShareSize - (len(csw.pendingShare.Share) - consts.NamespaceSize)
+	return len(csw.shares), availableBytes
+}
+
+// tail is filler for all tail padded shares
+// it is allocated once and used everywhere
+var tailPaddingShare = append(
+	append(make([]byte, 0, consts.ShareSize), consts.TailPaddingNamespaceID...),
+	bytes.Repeat([]byte{0}, consts.ShareSize-consts.NamespaceSize)...,
+)
+
+func TailPaddingShares(n int) NamespacedShares {
+	shares := make([]NamespacedShare, n)
+	for i := 0; i < n; i++ {
+		shares[i] = NamespacedShare{
+			Share: tailPaddingShare,
+			ID:    consts.TailPaddingNamespaceID,
+		}
+	}
+	return shares
+}
+
+func namespacedPaddedShares(ns []byte, count int) []NamespacedShare {
+	shares := make([]NamespacedShare, count)
+	for i := 0; i < count; i++ {
+		shares[i] = NamespacedShare{
+			Share: append(append(
+				make([]byte, 0, consts.ShareSize), ns...),
+				make([]byte, consts.MsgShareSize)...),
+			ID: ns,
+		}
+	}
+	return shares
+}
+
+func min(a, b int) int {
+	if a <= b {
+		return a
+	}
+	return b
+}
diff --git a/pkg/shares/split_message_shares.go b/pkg/shares/split_message_shares.go
new file mode 100644
index 0000000000..018707ab3c
--- /dev/null
+++ b/pkg/shares/split_message_shares.go
@@ -0,0 +1,134 @@
+package shares
+
+import (
+	"fmt"
+
+	"github.com/celestiaorg/nmt/namespace"
+	"github.com/tendermint/tendermint/pkg/consts"
+	coretypes "github.com/tendermint/tendermint/types"
+)
+
+// MessageShareSplitter lazily splits messages into shares that will eventually be
+// included in a data square. It also has methods to help progressively count
+// how many shares the messages written take up.
+type MessageShareSplitter struct {
+	shares [][]NamespacedShare
+	count  int
+}
+
+func NewMessageShareSplitter() *MessageShareSplitter {
+	return &MessageShareSplitter{}
+}
+
+// Write adds the delimited data to the underlying messages shares.
+func (msw *MessageShareSplitter) Write(msg coretypes.Message) {
+	rawMsg, err := msg.MarshalDelimited()
+	if err != nil {
+		panic(fmt.Sprintf("app accepted a Message that can not be encoded %#v", msg))
+	}
+	newShares := make([]NamespacedShare, 0)
+	newShares = AppendToShares(newShares, msg.NamespaceID, rawMsg)
+	msw.shares = append(msw.shares, newShares)
+	msw.count += len(newShares)
+}
+
+// RemoveMessage will remove a message from the underlying message state. If
+// there is namespaced padding after the message, then that is also removed.
+func (msw *MessageShareSplitter) RemoveMessage(i int) (int, error) {
+	j := 1
+	initialCount := msw.count
+	if len(msw.shares) > i+1 {
+		_, msgLen, err := ParseDelimiter(msw.shares[i+1][0].Share[consts.NamespaceSize:])
+		if err != nil {
+			return 0, err
+		}
+		// 0 means that there is padding after the share that we are about to
+		// remove. to remove this padding, we increase j by 1
+		// with the message
+		if msgLen == 0 {
+			j++
+			msw.count -= len(msw.shares[j])
+		}
+	}
+	msw.count -= len(msw.shares[i])
+	copy(msw.shares[i:], msw.shares[i+j:])
+	msw.shares = msw.shares[:len(msw.shares)-j]
+	return initialCount - msw.count, nil
+}
+
+// WriteNamespacedPaddedShares adds empty shares using the namespace of the last written share.
+// This is useful to follow the message layout rules. It assumes that at least
+// one share has already been written, if not it panics.
+func (msw *MessageShareSplitter) WriteNamespacedPaddedShares(count int) {
+	if len(msw.shares) == 0 {
+		panic("cannot write empty namespaced shares on an empty MessageShareSplitter")
+	}
+	if count == 0 {
+		return
+	}
+	lastMessage := msw.shares[len(msw.shares)-1]
+	msw.shares = append(msw.shares, namespacedPaddedShares(lastMessage[0].ID, count))
+	msw.count += count
+}
+
+// Export finalizes and returns the underlying contiguous shares.
+func (msw *MessageShareSplitter) Export() NamespacedShares {
+	shares := make([]NamespacedShare, msw.count)
+	cursor := 0
+	for _, messageShares := range msw.shares {
+		for _, share := range messageShares {
+			shares[cursor] = share
+			cursor++
+		}
+	}
+	return shares
+}
+
+// Count returns the current number of shares that will be made if exporting.
+func (msw *MessageShareSplitter) Count() int {
+	return msw.count
+}
+
+// AppendToShares appends raw data as shares.
+// Used for messages.
+func AppendToShares(shares []NamespacedShare, nid namespace.ID, rawData []byte) []NamespacedShare {
+	if len(rawData) <= consts.MsgShareSize {
+		rawShare := append(append(
+			make([]byte, 0, len(nid)+len(rawData)),
+			nid...),
+			rawData...,
+		)
+		paddedShare := zeroPadIfNecessary(rawShare, consts.ShareSize)
+		share := NamespacedShare{paddedShare, nid}
+		shares = append(shares, share)
+	} else { // len(rawData) > MsgShareSize
+		shares = append(shares, splitMessage(rawData, nid)...)
+	}
+	return shares
+}
+
+// splitMessage breaks the data in a message into the minimum number of
+// namespaced shares
+func splitMessage(rawData []byte, nid namespace.ID) NamespacedShares {
+	shares := make([]NamespacedShare, 0)
+	firstRawShare := append(append(
+		make([]byte, 0, consts.ShareSize),
+		nid...),
+		rawData[:consts.MsgShareSize]...,
+	)
+	shares = append(shares, NamespacedShare{firstRawShare, nid})
+	rawData = rawData[consts.MsgShareSize:]
+	for len(rawData) > 0 {
+		shareSizeOrLen := min(consts.MsgShareSize, len(rawData))
+		rawShare := append(append(
+			make([]byte, 0, consts.ShareSize),
+			nid...),
+			rawData[:shareSizeOrLen]...,
+		)
+		paddedShare := zeroPadIfNecessary(rawShare, consts.ShareSize)
+		share := NamespacedShare{paddedShare, nid}
+		shares = append(shares, share)
+		rawData = rawData[shareSizeOrLen:]
+	}
+	return shares
+}
diff --git a/pkg/shares/utils.go b/pkg/shares/utils.go
new file mode 100644
index 0000000000..e045e3c828
--- /dev/null
+++ b/pkg/shares/utils.go
@@ -0,0 +1,77 @@
+package shares
+
+import (
+	"math/bits"
+
+	"github.com/tendermint/tendermint/pkg/consts"
+	core "github.com/tendermint/tendermint/proto/tendermint/types"
+	coretypes "github.com/tendermint/tendermint/types"
+)
+
+// DelimLen calculates the length of the delimiter for a given message size
+func DelimLen(x uint64) int {
+	return 8 - bits.LeadingZeros64(x)%8
+}
+
+// MsgSharesUsed calculates the minimum number of shares a message will take up.
+// It accounts for the necessary delimiter and potential padding.
+func MsgSharesUsed(msgSize int) int {
+	// add the delimiter to the message size
+	msgSize = DelimLen(uint64(msgSize)) + msgSize
+	shareCount := msgSize / consts.MsgShareSize
+	// increment the share count if the message overflows the last counted share
+	if msgSize%consts.MsgShareSize != 0 {
+		shareCount++
+	}
+	return shareCount
+}
+
+func MessageShareCountsFromMessages(msgs []*core.Message) []int {
+	e := make([]int, len(msgs))
+	for i, msg := range msgs {
+		e[i] = MsgSharesUsed(len(msg.Data))
+	}
+	return e
+}
+
+func isPowerOf2(v uint64) bool {
+	return v&(v-1) == 0 && v != 0
+}
+
+func MessagesToProto(msgs []coretypes.Message) []*core.Message {
+	protoMsgs := make([]*core.Message, len(msgs))
+	for i, msg := range msgs {
+		protoMsgs[i] = &core.Message{
+			NamespaceId: msg.NamespaceID,
+			Data:        msg.Data,
+		}
+	}
+	return protoMsgs
+}
+
+func MessagesFromProto(msgs []*core.Message) []coretypes.Message {
+	protoMsgs := make([]coretypes.Message, len(msgs))
+	for i, msg := range msgs {
+		protoMsgs[i] = coretypes.Message{
+			NamespaceID: msg.NamespaceId,
+			Data:        msg.Data,
+		}
+	}
+	return protoMsgs
+}
+
+func TxsToBytes(txs coretypes.Txs) [][]byte {
+	e := make([][]byte, len(txs))
+	for i, tx := range txs {
+		e[i] = []byte(tx)
+	}
+	return e
+}
+
+func TxsFromBytes(txs [][]byte) coretypes.Txs {
+	e := make(coretypes.Txs, len(txs))
+	for i, tx := range txs {
+		e[i] = coretypes.Tx(tx)
+	}
+	return e
+}
diff --git a/x/payment/types/builder_test.go b/x/payment/types/builder_test.go
index 29ed23299c..269835d989 100644
--- a/x/payment/types/builder_test.go
+++ b/x/payment/types/builder_test.go
@@ -36,7 +36,7 @@ func TestBuildWirePayForData(t *testing.T) {
 	rawTx, err := makePaymentEncodingConfig().TxConfig.TxEncoder()(signedTx)
 	require.NoError(t, err)
 
-	_, _, isMalleated := coretypes.UnwrapMalleatedTx(rawTx)
+	_, isMalleated := coretypes.UnwrapMalleatedTx(rawTx)
 	require.False(t, isMalleated)
 
 	sigs, err := signedTx.GetSignaturesV2()
diff --git a/x/payment/types/payfordata.go b/x/payment/types/payfordata.go
index e251eb89a5..30b0d28b69 100644
--- a/x/payment/types/payfordata.go
+++ b/x/payment/types/payfordata.go
@@ -135,7 +135,10 @@ func CreateCommitment(k uint64, namespace, message []byte) ([]byte, error) {
 
 	// split into shares that are length delimited and include the namespace in
 	// each share
-	shares := shares.SplitMessagesIntoShares(msg).RawShares()
+	shares, err := shares.SplitMessages(nil, msg.MessagesList)
+	if err != nil {
+		return nil, err
+	}
 	// if the number of shares is larger than that in the square, throw an error
 	// note, we use k*k-1 here because at least a single share will be reserved
 	// for the transaction paying for the message, therefore the max number of