Skip to content

Commit a339289

Browse files
julienrbrtjulienrbrt
committed
revert hacks from #3290 
1 parent 3869cf9 commit a339289

2 files changed

Lines changed: 49 additions & 156 deletions

File tree

‎block/internal/da/fiber_client.go‎

Lines changed: 34 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import (
55
"encoding/binary"
66
"errors"
77
"fmt"
8-
"sync"
98
"time"
109

1110
"github.com/rs/zerolog"
@@ -88,96 +87,36 @@ func (c *fiberDAClient) Submit(ctx context.Context, data [][]byte, _ float64, na
8887
}
8988
}
9089

91-
// Per-item concurrent Upload. Fibre's per-Upload latency is
92-
// dominated by validator signature aggregation (~1.5 s on a
93-
// healthy network) and does not scale up linearly under multiple
94-
// in-flight Uploads, so settlement throughput scales with the
95-
// number of concurrent items submitted in a single batch. Each
96-
// item gets its own goroutine, its own Upload call, and its own
97-
// BlobID in the result; the previous flatten step was both
98-
// memory-wasteful (a MaxBlobSize-sized memcpy on every Submit)
99-
// and inherently serial (one Upload per Submit).
100-
//
101-
// TODO: wire-format compat — old splitBlobs assumed all items in
102-
// a Submit were written as a single prefixed blob. With per-item
103-
// Uploads, retrievers must treat each BlobID separately. The
104-
// retrieve path in this file still uses splitBlobs and will need
105-
// a follow-up to read the new per-item blobs as raw payloads.
90+
flat := flattenBlobs(data)
10691
nsID := namespace[len(namespace)-10:]
107-
type uploadResult struct {
108-
idx int
109-
id []byte
110-
err error
111-
}
112-
results := make([]uploadResult, len(data))
113-
var wg sync.WaitGroup
114-
for i := range data {
115-
wg.Add(1)
116-
go func(i int) {
117-
defer wg.Done()
118-
res, err := c.fiber.Upload(ctx, nsID, data[i])
119-
if err != nil {
120-
results[i] = uploadResult{idx: i, err: err}
121-
return
122-
}
123-
id := make([]byte, len(res.BlobID))
124-
copy(id, res.BlobID)
125-
results[i] = uploadResult{idx: i, id: id}
126-
}(i)
127-
}
128-
wg.Wait()
129-
130-
// Walk results in submission order. submitToDA's retry logic
131-
// expects "prefix of successes": SubmittedCount=N means items
132-
// [0..N) succeeded and the caller will re-submit items [N..end)
133-
// on the next attempt. Reporting interleaved successes would
134-
// double-submit blobs and waste escrow; matching prefix
135-
// semantics keeps the contract intact even when individual
136-
// Uploads fail out-of-order.
137-
ids := make([][]byte, 0, len(data))
138-
var firstErr error
139-
for _, r := range results {
140-
if r.err != nil {
141-
firstErr = r.err
142-
break
143-
}
144-
ids = append(ids, r.id)
145-
}
146-
147-
if len(ids) == 0 && firstErr != nil {
92+
result, err := c.fiber.Upload(context.Background(), nsID, flat)
93+
if err != nil {
14894
code := datypes.StatusError
14995
switch {
150-
case errors.Is(firstErr, context.Canceled):
96+
case errors.Is(err, context.Canceled):
15197
code = datypes.StatusContextCanceled
152-
case errors.Is(firstErr, context.DeadlineExceeded):
98+
case errors.Is(err, context.DeadlineExceeded):
15399
code = datypes.StatusContextDeadline
154100
}
155-
c.logger.Error().Err(firstErr).Msg("fiber upload failed")
101+
c.logger.Error().Err(err).Msg("fiber upload failed")
156102
return datypes.ResultSubmit{
157103
BaseResult: datypes.BaseResult{
158104
Code: code,
159-
Message: fmt.Sprintf("fiber upload failed for blob: %v", firstErr),
160-
SubmittedCount: 0,
105+
Message: fmt.Sprintf("fiber upload failed for blob: %v", err),
106+
SubmittedCount: uint64(len(data) - 1),
161107
BlobSize: blobSize,
162108
Timestamp: time.Now(),
163109
},
164110
}
165111
}
166112

167-
if firstErr != nil {
168-
c.logger.Warn().Err(firstErr).
169-
Int("submitted", len(ids)).
170-
Int("total", len(data)).
171-
Msg("fiber upload partial success — caller will retry the remainder")
172-
}
173-
174-
c.logger.Debug().Int("num_ids", len(ids)).Uint64("height", 0 /* TODO */).Msg("fiber DA submission successful")
113+
c.logger.Debug().Int("num_ids", len(data)).Uint64("height", 0 /* TODO */).Msg("fiber DA submission successful")
175114

176115
return datypes.ResultSubmit{
177116
BaseResult: datypes.BaseResult{
178117
Code: datypes.StatusSuccess,
179-
IDs: ids,
180-
SubmittedCount: uint64(len(ids)),
118+
IDs: [][]byte{result.BlobID},
119+
SubmittedCount: uint64(len(data)),
181120
Height: 0, /* TODO */
182121
BlobSize: blobSize,
183122
Timestamp: time.Now(),
@@ -398,19 +337,29 @@ func (c *fiberDAClient) Validate(_ context.Context, ids []datypes.ID, proofs []d
398337
func (c *fiberDAClient) GetHeaderNamespace() []byte { return c.namespaceBz }
399338
func (c *fiberDAClient) GetDataNamespace() []byte { return c.dataNamespaceBz }
400339

401-
// splitBlobs decodes the legacy "count + per-item length" framing that
402-
// the previous Submit path used to pack multiple blobs into a single
403-
// Upload. The per-item-concurrent Submit path no longer writes that
404-
// framing — each item is uploaded raw — so any blob written by this
405-
// branch's Submit will fail to decode here.
406-
//
407-
// Callers (Retrieve / RetrieveBlobs / Get / Subscribe) therefore only
408-
// work for blobs written by the OLD code path, OR for the multi-item
409-
// header batches that still use it. Pair the format change with a
410-
// matching update to the read path before any node on this branch
411-
// tries to sync from another node on this branch.
412-
//
413-
// Tracked alongside the wire-format TODO on Submit (above).
340+
func flattenBlobs(blobs [][]byte) []byte {
341+
if len(blobs) == 0 {
342+
return nil
343+
}
344+
345+
var total int
346+
for _, b := range blobs {
347+
total += 4 + len(b)
348+
}
349+
total += 4
350+
351+
buf := make([]byte, total)
352+
binary.BigEndian.PutUint32(buf, uint32(len(blobs)))
353+
off := 4
354+
for _, b := range blobs {
355+
binary.BigEndian.PutUint32(buf[off:], uint32(len(b)))
356+
off += 4
357+
copy(buf[off:], b)
358+
off += len(b)
359+
}
360+
return buf
361+
}
362+
414363
func splitBlobs(data []byte) ([][]byte, error) {
415364
if len(data) == 0 {
416365
return nil, nil

‎block/internal/submitting/da_submitter.go‎

Lines changed: 15 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -43,48 +43,14 @@ type retryPolicy struct {
4343
MinBackoff time.Duration
4444
MaxBackoff time.Duration
4545
MaxBlobBytes uint64
46-
// MaxItems caps the number of items packed into a single Submit
47-
// call. DA clients that fan out per-item Uploads (fiber) benefit
48-
// linearly from larger batches — settlement throughput scales
49-
// with concurrency until per-Upload latency dominates. Default
50-
// 1 preserves legacy single-item-per-Submit semantics for
51-
// backends that flatten a batch into one blob (JSON-RPC blob
52-
// client). The fiber path overrides this from config.
53-
MaxItems int
5446
}
5547

56-
// defaultBatchItems is the conservative default for non-fiber backends
57-
// that historically expected one item per Submit call. The fiber path
58-
// raises this via config because it can fan out per-item Uploads.
59-
const defaultBatchItems = 1
60-
61-
// fiberDefaultBatchItems is the upper bound on items packed into a
62-
// single fiber Submit. Each item gets its own concurrent Upload, so
63-
// this caps the per-batch goroutine fan-out.
64-
//
65-
// HACK(fiber-throughput): hardcoded at 16. The right value depends on
66-
// memory budget × per-item Upload size × Fibre validator-side
67-
// throughput, none of which the submitter can know at compile time.
68-
// Should be a config knob (FiberDAConfig.UploadConcurrency was
69-
// scaffolded for exactly this earlier — wire it through here when the
70-
// concurrent-uploads change graduates from prototype). 16 is a
71-
// pragmatic measurement default that gives meaningful concurrency
72-
// without overwhelming celestia-node's per-FSP rate.
73-
const fiberDefaultBatchItems = 16
74-
7548
func defaultRetryPolicy(maxAttempts int, maxDuration time.Duration) retryPolicy {
7649
return retryPolicy{
77-
MaxAttempts: maxAttempts,
78-
MinBackoff: initialBackoff,
79-
MaxBackoff: maxDuration,
80-
// TODO(throughput-cleanup): same value is used by
81-
// executing/executor.go::RetrieveBatch as the raw-tx budget
82-
// (with a 2% reservation) and again here as the marshaled
83-
// blob ceiling. They are semantically different limits;
84-
// the duplication is what made packed-block-larger-than-cap
85-
// failures non-obvious. See common/consts.go.
50+
MaxAttempts: maxAttempts,
51+
MinBackoff: initialBackoff,
52+
MaxBackoff: maxDuration,
8653
MaxBlobBytes: common.DefaultMaxBlobSize,
87-
MaxItems: defaultBatchItems,
8854
}
8955
}
9056

@@ -606,19 +572,12 @@ func submitToDA[T any](
606572
}
607573

608574
pol := defaultRetryPolicy(s.config.DA.MaxSubmitAttempts, s.config.DA.BlockTime.Duration)
609-
// Fiber's DA client fans out per-item Uploads concurrently, so
610-
// packing more items per Submit lifts settlement throughput. For
611-
// non-fiber backends the default of 1 preserves the legacy
612-
// flatten-one-blob behavior.
613-
if s.config.DA.IsFiberEnabled() {
614-
pol.MaxItems = fiberDefaultBatchItems
615-
}
616575

617576
rs := retryState{Attempt: 0, Backoff: 0}
618577

619578
// Limit this submission to a single size-capped batch
620579
if len(marshaled) > 0 {
621-
batchItems, batchMarshaled, err := limitBatchBySize(items, marshaled, pol.MaxBlobBytes, pol.MaxItems)
580+
batchItems, batchMarshaled, err := limitBatchBySize(items, marshaled, pol.MaxBlobBytes)
622581
if err != nil {
623582
s.logger.Error().
624583
Str("itemType", itemType).
@@ -729,42 +688,27 @@ func submitToDA[T any](
729688
return fmt.Errorf("failed to submit all %s(s) to DA layer after %d attempts", itemType, rs.Attempt)
730689
}
731690

732-
// limitBatchBySize returns a prefix of items whose per-item marshaled size
733-
// fits within maxItemBytes. The total batch size is bounded by item count
734-
// (maxItems), not by total bytes — DA clients that can fan out per-item
735-
// Uploads (e.g. the fiber DA client) settle each item in its own
736-
// concurrent Upload call, so packing more items per batch lifts the
737-
// effective settlement throughput. DA clients that flatten a batch into
738-
// a single blob still get one item per call when maxItems == 1.
739-
//
740-
// If the first item exceeds maxItemBytes, returns ErrOversizedItem
741-
// (unrecoverable). If no items fit at all (empty inputs), returns a
742-
// distinct error so the caller can distinguish "nothing to send".
743-
//
744-
// TODO(throughput-cleanup): see common/consts.go — maxItemBytes is the
745-
// per-item chain ceiling, separate from the raw-tx budget driving
746-
// FilterTxs. Once that split lands, the duplicate-cap-everywhere
747-
// problem these fixes work around goes away.
748-
func limitBatchBySize[T any](items []T, marshaled [][]byte, maxItemBytes uint64, maxItems int) ([]T, [][]byte, error) {
749-
if maxItems <= 0 {
750-
maxItems = 1
751-
}
691+
// limitBatchBySize returns a prefix of items whose total marshaled size does not exceed maxBytes.
692+
// If the first item exceeds maxBytes, it returns ErrOversizedItem which is unrecoverable.
693+
func limitBatchBySize[T any](items []T, marshaled [][]byte, maxBytes uint64) ([]T, [][]byte, error) {
694+
total := uint64(0)
752695
count := 0
753696
for i := range items {
754-
if count >= maxItems {
755-
break
756-
}
757697
sz := uint64(len(marshaled[i]))
758-
if sz > maxItemBytes {
698+
if sz > maxBytes {
759699
if i == 0 {
760-
return nil, nil, fmt.Errorf("%w: item size %d exceeds max %d", common.ErrOversizedItem, sz, maxItemBytes)
700+
return nil, nil, fmt.Errorf("%w: item size %d exceeds max %d", common.ErrOversizedItem, sz, maxBytes)
761701
}
762702
break
763703
}
704+
if total+sz > maxBytes {
705+
break
706+
}
707+
total += sz
764708
count++
765709
}
766710
if count == 0 {
767-
return nil, nil, fmt.Errorf("no items fit within %d bytes", maxItemBytes)
711+
return nil, nil, fmt.Errorf("no items fit within %d bytes", maxBytes)
768712
}
769713
return items[:count], marshaled[:count], nil
770714
}

0 commit comments

Comments
 (0)