-
Notifications
You must be signed in to change notification settings - Fork 176
/
serializer.go
205 lines (167 loc) · 5.85 KB
/
serializer.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
package execution_data
import (
"fmt"
"io"
"math"
cborlib "github.com/fxamacker/cbor/v2"
"github.com/ipfs/go-cid"
"github.com/onflow/flow-go/model/encoding"
"github.com/onflow/flow-go/model/encoding/cbor"
"github.com/onflow/flow-go/model/flow"
"github.com/onflow/flow-go/module/executiondatasync/execution_data/internal"
"github.com/onflow/flow-go/network"
"github.com/onflow/flow-go/network/compressor"
)
// DefaultSerializer is the default implementation for an Execution Data serializer.
// It is configured to use cbor encoding with LZ4 compression.
var DefaultSerializer Serializer
func init() {
var codec encoding.Codec
decMode, err := cborlib.DecOptions{
MaxArrayElements: math.MaxInt64,
MaxMapPairs: math.MaxInt64,
MaxNestedLevels: math.MaxInt16,
}.DecMode()
if err != nil {
panic(err)
}
codec = cbor.NewCodec(cbor.WithDecMode(decMode))
DefaultSerializer = NewSerializer(codec, compressor.NewLz4Compressor())
}
// header codes are used to distinguish between the different data types serialized within a blob.
// they provide simple versioning of execution state data blobs and indicate how the data should
// be deserialized back into their original form. Therefore, each input format must have a unique
// code, and the codes must never be reused. This allows libraries to accurately decode the data
// without juggling software versions.
const (
codeRecursiveCIDs = iota + 1
codeExecutionDataRoot
codeChunkExecutionDataV1
codeChunkExecutionDataV2 // includes transaction results
)
// getCode returns the header code for the given value's type.
// It returns an error if the type is not supported.
func getCode(v interface{}) (byte, error) {
switch v.(type) {
case *flow.BlockExecutionDataRoot:
return codeExecutionDataRoot, nil
case *internal.ChunkExecutionDataV1: // only used for backwards compatibility testing
return codeChunkExecutionDataV1, nil
case *ChunkExecutionData:
return codeChunkExecutionDataV2, nil
case []cid.Cid:
return codeRecursiveCIDs, nil
default:
return 0, fmt.Errorf("invalid type for interface: %T", v)
}
}
// getPrototype returns a new instance of the type that corresponds to the given header code.
// It returns an error if the code is not supported.
func getPrototype(code byte) (interface{}, error) {
switch code {
case codeExecutionDataRoot:
return &flow.BlockExecutionDataRoot{}, nil
case codeChunkExecutionDataV2, codeChunkExecutionDataV1:
return &ChunkExecutionData{}, nil // only return the latest version
case codeRecursiveCIDs:
return &[]cid.Cid{}, nil
default:
return nil, fmt.Errorf("invalid code: %v", code)
}
}
// Serializer is used to serialize / deserialize Execution Data and CID lists for the
// Execution Data Service.
type Serializer interface {
// Serialize encodes and compresses the given value to the given writer.
// No errors are expected during normal operation.
Serialize(io.Writer, interface{}) error
// Deserialize decompresses and decodes the data from the given reader.
// No errors are expected during normal operation.
Deserialize(io.Reader) (interface{}, error)
}
// serializer implements the Serializer interface. Object are serialized by encoding and
// compressing them using the given codec and compressor.
//
// The serialized data is prefixed with a single byte header that identifies the underlying
// data format. This allows adding new data types in a backwards compatible way.
type serializer struct {
codec encoding.Codec
compressor network.Compressor
}
// NewSerializer returns a new Execution Data serializer using the provided encoder and compressor.
func NewSerializer(codec encoding.Codec, compressor network.Compressor) *serializer {
return &serializer{
codec: codec,
compressor: compressor,
}
}
// writePrototype writes the header code for the given value to the given writer
func (s *serializer) writePrototype(w io.Writer, v interface{}) error {
var code byte
var err error
if code, err = getCode(v); err != nil {
return err
}
if bw, ok := w.(io.ByteWriter); ok {
err = bw.WriteByte(code)
} else {
_, err = w.Write([]byte{code})
}
if err != nil {
return fmt.Errorf("failed to write code: %w", err)
}
return nil
}
// Serialize encodes and compresses the given value to the given writer.
// No errors are expected during normal operation.
func (s *serializer) Serialize(w io.Writer, v interface{}) error {
if err := s.writePrototype(w, v); err != nil {
return fmt.Errorf("failed to write prototype: %w", err)
}
comp, err := s.compressor.NewWriter(w)
if err != nil {
return fmt.Errorf("failed to create compressor writer: %w", err)
}
enc := s.codec.NewEncoder(comp)
if err := enc.Encode(v); err != nil {
return fmt.Errorf("failed to encode data: %w", err)
}
// flush data out to the underlying writer
if err := comp.Close(); err != nil {
return fmt.Errorf("failed to close compressor: %w", err)
}
return nil
}
// readPrototype reads a header code from the given reader and returns a prototype value
func (s *serializer) readPrototype(r io.Reader) (interface{}, error) {
var code byte
var err error
if br, ok := r.(io.ByteReader); ok {
code, err = br.ReadByte()
} else {
var buf [1]byte
_, err = r.Read(buf[:])
code = buf[0]
}
if err != nil {
return nil, fmt.Errorf("failed to read code: %w", err)
}
return getPrototype(code)
}
// Deserialize decompresses and decodes the data from the given reader.
// No errors are expected during normal operation.
func (s *serializer) Deserialize(r io.Reader) (interface{}, error) {
v, err := s.readPrototype(r)
if err != nil {
return nil, fmt.Errorf("failed to read prototype: %w", err)
}
comp, err := s.compressor.NewReader(r)
if err != nil {
return nil, fmt.Errorf("failed to create compressor reader: %w", err)
}
dec := s.codec.NewDecoder(comp)
if err := dec.Decode(v); err != nil {
return nil, fmt.Errorf("failed to decode data: %w", err)
}
return v, nil
}