this repo has no description
add 'too big' event flag to events (#62)
We don't have a lexicon for this yet, but this does prevent things from
breaking outright for now.
authored by
Whyrusleeping
and committed by
GitHub
216879fc
aef25f3a
+75
-12
+1
-1
bgs/bgs.go
+1
-1
bgs/bgs.go
+2
-5
carstore/bs.go
+2
-5
carstore/bs.go
···
330
330
331
331
if earlyCid.Defined() {
332
332
var untilShard CarShard
333
-
if err := cs.meta.First(&untilShard, "root = ? AND usr = ?", earlyCid.String(), user).Error; err != nil {
333
+
if err := cs.meta.First(&untilShard, "root = ? AND usr = ?", util.DbCID{earlyCid}, user).Error; err != nil {
334
334
return err
335
335
}
336
336
earlySeq = untilShard.Seq
···
338
338
339
339
if lateCid.Defined() {
340
340
var fromShard CarShard
341
-
if err := cs.meta.First(&fromShard, "root = ? AND usr = ?", lateCid.String(), user).Error; err != nil {
341
+
if err := cs.meta.First(&fromShard, "root = ? AND usr = ?", util.DbCID{lateCid}, user).Error; err != nil {
342
342
return err
343
343
}
344
344
lateSeq = fromShard.Seq
···
529
529
return nil, err
530
530
}
531
531
532
-
if buf.Len() > MaxSliceLength {
533
-
return nil, fmt.Errorf("cannot close carstore session, too much data written (%d)", buf.Len())
534
-
}
535
532
/*
536
533
brefs = append(brefs, &blockRef{
537
534
Cid: k.String(),
+2
-1
cmd/bigsky/main.go
+2
-1
cmd/bigsky/main.go
+16
-1
cmd/stress/main.go
+16
-1
cmd/stress/main.go
···
76
76
Value: "http://localhost:4849",
77
77
EnvVars: []string{"ATP_PDS_HOST"},
78
78
},
79
+
&cli.StringFlag{
80
+
Name: "invite",
81
+
},
79
82
},
80
83
Action: func(cctx *cli.Context) error {
81
84
xrpcc, err := cliutil.GetXrpcClient(cctx, false)
···
93
96
id := hex.EncodeToString(buf)
94
97
95
98
var invite *string
99
+
if inv := cctx.String("invite"); inv != "" {
100
+
invite = &inv
101
+
}
102
+
103
+
cfg, err := comatproto.ServerGetAccountsConfig(ctx, xrpcc)
104
+
if err != nil {
105
+
return err
106
+
}
107
+
108
+
domain := cfg.AvailableUserDomains[0]
109
+
fmt.Println("domain: ", domain)
110
+
96
111
resp, err := comatproto.AccountCreate(ctx, xrpcc, &comatproto.AccountCreate_Input{
97
112
Email: fmt.Sprintf("user-%s@test.com", id),
98
-
Handle: "user-" + id + ".test",
113
+
Handle: "user-" + id + domain,
99
114
Password: "password",
100
115
InviteCode: invite,
101
116
})
+35
-1
events/cbor_gen.go
+35
-1
events/cbor_gen.go
···
135
135
136
136
cw := cbg.NewCborWriter(w)
137
137
138
-
if _, err := cw.Write([]byte{169}); err != nil {
138
+
if _, err := cw.Write([]byte{170}); err != nil {
139
139
return err
140
140
}
141
141
···
360
360
return err
361
361
}
362
362
if _, err := io.WriteString(w, string(t.Commit)); err != nil {
363
+
return err
364
+
}
365
+
366
+
// t.TooBig (bool) (bool)
367
+
if len("tooBig") > cbg.MaxLength {
368
+
return xerrors.Errorf("Value in field \"tooBig\" was too long")
369
+
}
370
+
371
+
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("tooBig"))); err != nil {
372
+
return err
373
+
}
374
+
if _, err := io.WriteString(w, string("tooBig")); err != nil {
375
+
return err
376
+
}
377
+
378
+
if err := cbg.WriteBool(w, t.TooBig); err != nil {
363
379
return err
364
380
}
365
381
return nil
···
577
593
}
578
594
579
595
t.Commit = string(sval)
596
+
}
597
+
// t.TooBig (bool) (bool)
598
+
case "tooBig":
599
+
600
+
maj, extra, err = cr.ReadHeader()
601
+
if err != nil {
602
+
return err
603
+
}
604
+
if maj != cbg.MajOther {
605
+
return fmt.Errorf("booleans must be major type 7")
606
+
}
607
+
switch extra {
608
+
case 20:
609
+
t.TooBig = false
610
+
case 21:
611
+
t.TooBig = true
612
+
default:
613
+
return fmt.Errorf("booleans are either major type 7, value 20 or 21 (got %d)", extra)
580
614
}
581
615
582
616
default:
+5
-1
events/dbpersist.go
+5
-1
events/dbpersist.go
+10
-2
events/events.go
+10
-2
events/events.go
···
2
2
3
3
import (
4
4
"context"
5
+
"errors"
5
6
"fmt"
6
7
7
8
"github.com/bluesky-social/indigo/util"
···
121
122
122
123
Ops []*RepoOp `cborgen:"ops"`
123
124
Blocks []byte `cborgen:"blocks"`
125
+
TooBig bool `cborgen:"tooBig"`
124
126
125
127
Blobs []string `cborgen:"blobs"`
126
128
···
175
177
}
176
178
}
177
179
180
+
var ErrPlaybackShutdown = fmt.Errorf("playback shutting down")
181
+
178
182
func (em *EventManager) Subscribe(ctx context.Context, filter func(*XRPCStreamEvent) bool, since *int64) (<-chan *XRPCStreamEvent, func(), error) {
179
183
if filter == nil {
180
184
filter = func(*XRPCStreamEvent) bool { return true }
···
192
196
if err := em.persister.Playback(ctx, *since, func(e *XRPCStreamEvent) error {
193
197
select {
194
198
case <-done:
195
-
return fmt.Errorf("shutting down")
199
+
return ErrPlaybackShutdown
196
200
case sub.outgoing <- e:
197
201
return nil
198
202
}
199
203
}); err != nil {
200
-
log.Errorf("events playback: %s", err)
204
+
if errors.Is(err, ErrPlaybackShutdown) {
205
+
log.Warnf("events playback: %s", err)
206
+
} else {
207
+
log.Errorf("events playback: %s", err)
208
+
}
201
209
}
202
210
}
203
211