Skip to content

Adding ordinals to actions stream + trace read fixes #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
aaroncox merged 2 commits into from
Jun 6, 2026
Merged

Adding ordinals to actions stream + trace read fixes #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
42fd5e1
handle trace live edge reads and enforce gap-free block writes
aaroncox Jun 6, 2026
9728b81
stream action ordinals end-to-end; fix authorizer-indexed catchup, sp…
aaroncox Jun 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 37 additions & 28 deletions libraries/actionstream/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,13 +20,13 @@ var (
)

const (
MsgTypeActionSubscribe uint8 = 0x30
MsgTypeActionAck uint8 = 0x31
MsgTypeActionBatch uint8 = 0x32
MsgTypeActionHeartbeat uint8 = 0x33
MsgTypeActionError uint8 = 0x34
MsgTypeActionDecoded uint8 = 0x35
MsgTypeCatchupComplete uint8 = 0x36
MsgTypeActionSubscribe uint8 = 0x30
MsgTypeActionAck uint8 = 0x31
MsgTypeActionBatch uint8 = 0x32
MsgTypeActionHeartbeat uint8 = 0x33
MsgTypeActionError uint8 = 0x34
MsgTypeActionDecoded uint8 = 0x35
MsgTypeCatchupComplete uint8 = 0x36

MaxMessageSize = 10 * 1024 * 1024
)
Expand All @@ -51,15 +51,18 @@ func DefaultClientConfig() ClientConfig {
}

type Action struct {
GlobalSeq uint64
BlockNum uint32
BlockTime uint32
Contract string
Action string
Receiver string
ActionData []byte
CpuUsageUs uint32
NetUsageWords uint32
GlobalSeq uint64
BlockNum uint32
BlockTime uint32
Contract string
Action string
Receiver string
ActionData []byte
CpuUsageUs uint32
NetUsageWords uint32
ActionOrdinal uint32
CreatorActionOrdinal uint32
ClosestUnnotifiedAncestorActionOrdinal uint32
}

type Filter struct {
Expand Down Expand Up @@ -492,7 +495,7 @@ func (c *Client) writeMessage(w io.Writer, msgType uint8, payload []byte) error
}

func (c *Client) decodeAction(payload []byte) (Action, error) {
if len(payload) < 48 {
if len(payload) < 60 {
return Action{}, errors.New("action payload too short")
}

Expand All @@ -504,22 +507,28 @@ func (c *Client) decodeAction(payload []byte) (Action, error) {
receiver := binary.LittleEndian.Uint64(payload[32:40])
cpuUsageUs := binary.LittleEndian.Uint32(payload[40:44])
netUsageWords := binary.LittleEndian.Uint32(payload[44:48])
actionOrdinal := binary.LittleEndian.Uint32(payload[48:52])
creatorActionOrdinal := binary.LittleEndian.Uint32(payload[52:56])
closestUAAO := binary.LittleEndian.Uint32(payload[56:60])

var actionData []byte
if len(payload) > 48 {
actionData = payload[48:]
if len(payload) > 60 {
actionData = payload[60:]
}

return Action{
GlobalSeq: globalSeq,
BlockNum: blockNum,
BlockTime: blockTime,
Contract: chain.NameToString(contract),
Action: chain.NameToString(actionName),
Receiver: chain.NameToString(receiver),
ActionData: actionData,
CpuUsageUs: cpuUsageUs,
NetUsageWords: netUsageWords,
GlobalSeq: globalSeq,
BlockNum: blockNum,
BlockTime: blockTime,
Contract: chain.NameToString(contract),
Action: chain.NameToString(actionName),
Receiver: chain.NameToString(receiver),
ActionData: actionData,
CpuUsageUs: cpuUsageUs,
NetUsageWords: netUsageWords,
ActionOrdinal: actionOrdinal,
CreatorActionOrdinal: creatorActionOrdinal,
ClosestUnnotifiedAncestorActionOrdinal: closestUAAO,
}, nil
}

Expand Down
51 changes: 44 additions & 7 deletions libraries/actionstream/client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ func TestNewClient(t *testing.T) {
func TestDecodeAction(t *testing.T) {
client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig())

payload := make([]byte, 48)
payload := make([]byte, 60)
binary.LittleEndian.PutUint64(payload[0:8], 12345)
binary.LittleEndian.PutUint32(payload[8:12], 1000)
binary.LittleEndian.PutUint32(payload[12:16], 1700000000)
Expand All @@ -37,6 +37,9 @@ func TestDecodeAction(t *testing.T) {
binary.LittleEndian.PutUint64(payload[32:40], 0x5530EA033C80A555)
binary.LittleEndian.PutUint32(payload[40:44], 500)
binary.LittleEndian.PutUint32(payload[44:48], 20)
binary.LittleEndian.PutUint32(payload[48:52], 0)
binary.LittleEndian.PutUint32(payload[52:56], 0)
binary.LittleEndian.PutUint32(payload[56:60], 0)

action, err := client.decodeAction(payload)
if err != nil {
Expand All @@ -63,7 +66,7 @@ func TestDecodeAction(t *testing.T) {
func TestDecodeActionWithData(t *testing.T) {
client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig())

payload := make([]byte, 56)
payload := make([]byte, 68)
binary.LittleEndian.PutUint64(payload[0:8], 100)
binary.LittleEndian.PutUint32(payload[8:12], 500)
binary.LittleEndian.PutUint32(payload[12:16], 1600000000)
Expand All @@ -72,7 +75,10 @@ func TestDecodeActionWithData(t *testing.T) {
binary.LittleEndian.PutUint64(payload[32:40], 0)
binary.LittleEndian.PutUint32(payload[40:44], 0)
binary.LittleEndian.PutUint32(payload[44:48], 0)
copy(payload[48:], []byte("testdata"))
binary.LittleEndian.PutUint32(payload[48:52], 0)
binary.LittleEndian.PutUint32(payload[52:56], 0)
binary.LittleEndian.PutUint32(payload[56:60], 0)
copy(payload[60:], []byte("testdata"))

action, err := client.decodeAction(payload)
if err != nil {
Expand All @@ -87,10 +93,41 @@ func TestDecodeActionWithData(t *testing.T) {
func TestDecodeActionTooShort(t *testing.T) {
client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig())

payload := make([]byte, 30)
payload := make([]byte, 48)
_, err := client.decodeAction(payload)
if err == nil {
t.Error("expected error for short payload")
t.Error("expected error for payload shorter than 60 bytes")
}
}

func TestDecodeActionOrdinals(t *testing.T) {
client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig())

build := func(ao, creator, closest uint32) []byte {
p := make([]byte, 60)
binary.LittleEndian.PutUint64(p[0:8], 7)
binary.LittleEndian.PutUint32(p[48:52], ao)
binary.LittleEndian.PutUint32(p[52:56], creator)
binary.LittleEndian.PutUint32(p[56:60], closest)
return p
}

topLevel, err := client.decodeAction(build(1, 0, 0))
if err != nil {
t.Fatalf("decodeAction (top-level) failed: %v", err)
}
if topLevel.ActionOrdinal != 1 || topLevel.CreatorActionOrdinal != 0 || topLevel.ClosestUnnotifiedAncestorActionOrdinal != 0 {
t.Errorf("top-level ordinals = (%d,%d,%d), want (1,0,0)",
topLevel.ActionOrdinal, topLevel.CreatorActionOrdinal, topLevel.ClosestUnnotifiedAncestorActionOrdinal)
}

inline, err := client.decodeAction(build(3, 1, 1))
if err != nil {
t.Fatalf("decodeAction (inline) failed: %v", err)
}
if inline.ActionOrdinal != 3 || inline.CreatorActionOrdinal != 1 || inline.ClosestUnnotifiedAncestorActionOrdinal != 1 {
t.Errorf("inline ordinals = (%d,%d,%d), want (3,1,1)",
inline.ActionOrdinal, inline.CreatorActionOrdinal, inline.ClosestUnnotifiedAncestorActionOrdinal)
}
}

Expand Down Expand Up @@ -644,7 +681,7 @@ func TestDialReceivesBatch(t *testing.T) {
payload := make([]byte, length-1)
conn.Read(payload)

actionPayload := make([]byte, 48)
actionPayload := make([]byte, 60)
binary.LittleEndian.PutUint64(actionPayload[0:8], 999)
binary.LittleEndian.PutUint32(actionPayload[8:12], 100)
binary.LittleEndian.PutUint32(actionPayload[12:16], 1700000000)
Expand Down Expand Up @@ -832,7 +869,7 @@ func TestRecvLoopReceivesActions(t *testing.T) {
close(serverReady)

for i := uint64(0); i < 3; i++ {
actionPayload := make([]byte, 48)
actionPayload := make([]byte, 60)
binary.LittleEndian.PutUint64(actionPayload[0:8], 1000+i)
binary.LittleEndian.PutUint32(actionPayload[8:12], uint32(100+i))
binary.LittleEndian.PutUint32(actionPayload[12:16], 1700000000)
Expand Down
55 changes: 19 additions & 36 deletions libraries/corereader/canonical_filter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,21 @@ type combinedActionInfo struct {
TrxIndex uint32
}

func newFilteredAction(account uint64, info *combinedActionInfo, globalSeq uint64, isAuth bool) Action {
return Action{
Account: account,
Contract: info.Meta.Contract,
Action: info.Meta.Action,
GlobalSeq: globalSeq,
TrxIndex: info.TrxIndex,
IsAuthorizer: isAuth,
Receiver: info.Receiver,
ActionOrdinal: info.Action.ActionOrdinal,
CreatorAO: info.Action.CreatorAO,
ClosestUAAO: info.Action.ClosestUAAO,
}
}

type blockFilter struct {
// Slice-backed lookup: avoids struct copy on every map access
infoSlice []combinedActionInfo
Expand Down Expand Up @@ -315,15 +330,7 @@ func filterBlockInto(bf *blockFilter, notif RawBlock, actionFilter ActionFilterF
}
}
}
actionsBuf = append(actionsBuf, Action{
Account: account,
Contract: info.Meta.Contract,
Action: info.Meta.Action,
GlobalSeq: globalSeq,
TrxIndex: info.TrxIndex,
IsAuthorizer: isAuth,
Receiver: info.Receiver,
})
actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, isAuth))
if minSeq == 0 || globalSeq < minSeq {
minSeq = globalSeq
}
Expand Down Expand Up @@ -371,15 +378,7 @@ func filterBlockInto(bf *blockFilter, notif RawBlock, actionFilter ActionFilterF
}
}

actionsBuf = append(actionsBuf, Action{
Account: account,
Contract: info.Meta.Contract,
Action: info.Meta.Action,
GlobalSeq: globalSeq,
TrxIndex: info.TrxIndex,
IsAuthorizer: true,
Receiver: info.Receiver,
})
actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, true))
if minSeq == 0 || globalSeq < minSeq {
minSeq = globalSeq
}
Expand Down Expand Up @@ -499,15 +498,7 @@ func filterBlockIntoTimed(bf *blockFilter, notif RawBlock, actionFilter ActionFi
}
}
}
actionsBuf = append(actionsBuf, Action{
Account: account,
Contract: info.Meta.Contract,
Action: info.Meta.Action,
GlobalSeq: globalSeq,
TrxIndex: info.TrxIndex,
IsAuthorizer: isAuth,
Receiver: info.Receiver,
})
actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, isAuth))
if minSeq == 0 || globalSeq < minSeq {
minSeq = globalSeq
}
Expand Down Expand Up @@ -558,15 +549,7 @@ func filterBlockIntoTimed(bf *blockFilter, notif RawBlock, actionFilter ActionFi
}
}

actionsBuf = append(actionsBuf, Action{
Account: account,
Contract: info.Meta.Contract,
Action: info.Meta.Action,
GlobalSeq: globalSeq,
TrxIndex: info.TrxIndex,
IsAuthorizer: true,
Receiver: info.Receiver,
})
actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, true))
if minSeq == 0 || globalSeq < minSeq {
minSeq = globalSeq
}
Expand Down
43 changes: 43 additions & 0 deletions libraries/corereader/canonical_filter_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,49 @@ import (
"testing"
)

func TestFilterRawBlock_CarriesOrdinals(t *testing.T) {
namesInBlock := []uint64{1111, 5555}

notif := RawBlock{
BlockNum: 100,
BlockTime: 1000000,
NamesInBlock: namesInBlock,
Notifications: map[uint64][]uint64{
1111: {1},
},
ActionMeta: []ActionMetadata{
{GlobalSeq: 1, Contract: 5555, Action: 6666},
},
Actions: []CanonicalAction{
{
ActionOrdinal: 3,
CreatorAO: 1,
ClosestUAAO: 1,
ReceiverUint64: 5555,
DataIndex: 0,
AuthAccountIndexes: []uint32{0},
GlobalSeqUint64: 1,
TrxIndex: 0,
ContractUint64: 5555,
ActionUint64: 6666,
},
},
}

filtered := FilterRawBlock(notif, nil)
if len(filtered.Actions) == 0 {
t.Fatal("expected filtered actions, got 0")
}
for _, a := range filtered.Actions {
if a.GlobalSeq != 1 {
continue
}
if a.ActionOrdinal != 3 || a.CreatorAO != 1 || a.ClosestUAAO != 1 {
t.Errorf("seq 1 ordinals = (%d,%d,%d), want (3,1,1)", a.ActionOrdinal, a.CreatorAO, a.ClosestUAAO)
}
}
}

func TestFilterRawBlock_Actions(t *testing.T) {
namesInBlock := []uint64{1111, 2222, 5555, 8888}

Expand Down
4 changes: 3 additions & 1 deletion libraries/corereader/shared.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,7 @@ type syncBlockData struct {
type syncActionInfo struct {
ActionOrdinal uint32
CreatorAO uint32
ClosestUAAO uint32
ReceiverIndex uint32
ContractNameIndex uint32
ActionNameIndex uint32
Expand Down Expand Up @@ -537,8 +538,9 @@ func parseSyncActionDirect(r *byteSliceReader, blob *syncBlockBlob, authBufOffse
action.CreatorAO = uint32(r.ReadUvarint())
}

action.ClosestUAAO = action.CreatorAO
if (magicmask1 & (1 << 3)) == 0 {
r.ReadUvarint()
action.ClosestUAAO = action.CreatorAO ^ uint32(r.ReadUvarint())
}

if (magicmask1 & (1 << 2)) == 0 {
Expand Down
Loading
Loading