tree.go raw
1 // Package relay wires the relay's domains into the fixed spawn tree. The Tree
2 // value is the orchestrator: it creates channels and breakers, spawns the
3 // domains, waits for their startup, and tears the tree down. It holds no
4 // protocol, storage or socket state of its own.
5 package relay
6
7 import (
8 "errors"
9 "fmt"
10 "sync"
11 "time"
12
13 "git.smesh.lol/torly/pkg/config"
14 "git.smesh.lol/torly/pkg/domain/blob"
15 "git.smesh.lol/torly/pkg/domain/db"
16 iodomain "git.smesh.lol/torly/pkg/domain/io"
17 netdomain "git.smesh.lol/torly/pkg/domain/net"
18 "git.smesh.lol/torly/pkg/domain/verify"
19 tinyspawn "git.smesh.lol/torly/pkg/tinyspawn"
20 "git.smesh.lol/torly/pkg/wire"
21 )
22
23 // Tree is a running relay.
24 type Tree struct {
25 cfg config.Config
26 addr string
27 stop tinyspawn.Breaker
28 done []<-chan struct{}
29 once sync.Once
30 }
31
32 // Start opens the listening socket, spawns the domain tree and waits for the
33 // domains that own durable state to finish opening it.
34 func Start(cfg config.Config) (*Tree, error) {
35 sk, err := loadRelayKey(&cfg)
36 if err != nil {
37 return nil, fmt.Errorf("relay identity: %w", err)
38 }
39 t := &Tree{cfg: cfg, addr: cfg.Addr(), stop: tinyspawn.NewBreaker()}
40
41 dbReady := make(chan wire.Ready, 1)
42 blobReady := make(chan wire.Ready, 1)
43
44 dbReqNet, dbReqDB := tinyspawn.NewBufferedDuplex[wire.DBRequest](64)
45 dbRespNet, dbRespDB := tinyspawn.NewBufferedDuplex[wire.DBResponse](64)
46 blobReqNet, blobReqBlob := tinyspawn.NewBufferedDuplex[wire.BlobRequest](32)
47 blobRespNet, blobRespBlob := tinyspawn.NewBufferedDuplex[wire.BlobResponse](32)
48
49 t.add("db", tinyspawn.Spawn(db.Run, db.Data{
50 Dir: cfg.DataDir + "/db",
51 MaxResults: cfg.QueryLimit,
52 Expiry: time.Duration(cfg.ExpiryInterval) * time.Second,
53 }, db.Chans{Req: dbReqDB, Resp: dbRespDB, Ready: dbReady, Stop: t.stop}))
54
55 t.add("blob", tinyspawn.Spawn(blob.Run, blob.Data{
56 Dir: cfg.DataDir,
57 MaxBlob: cfg.BlossomMaxBlob,
58 }, blob.Chans{Req: blobReqBlob, Resp: blobRespBlob, Ready: blobReady, Stop: t.stop}))
59
60 if err := waitReadyFor(dbReady, 2*time.Minute); err != nil {
61 t.Close()
62 return nil, fmt.Errorf("database: %w", err)
63 }
64 if cfg.BlossomEnabled {
65 if err := waitReady(blobReady); err != nil {
66 t.Close()
67 return nil, fmt.Errorf("blossom: %w", err)
68 }
69 } else {
70 // The blob domain still opened an index; it is harmless and keeps the
71 // fixed tree shape.
72 _ = waitReady(blobReady)
73 }
74
75 netCh := netdomain.Chans{
76 DBReq: dbReqNet,
77 DBResp: dbRespNet,
78 BlobReq: blobReqNet,
79 BlobResp: blobRespNet,
80 Ready: make(chan wire.Ready, 1),
81 Stop: t.stop,
82 }
83
84 for i := 0; i < cfg.VerifyWorkers; i++ {
85 reqNet, reqWorker := tinyspawn.NewBufferedDuplex[wire.VerifyRequest](8)
86 respNet, respWorker := tinyspawn.NewBufferedDuplex[wire.VerifyResponse](8)
87 netCh.Verifiers = append(netCh.Verifiers, reqNet)
88 netCh.VerifyResp = append(netCh.VerifyResp, respNet)
89 t.add("verify", tinyspawn.Spawn(verify.Run, struct{}{}, verify.Chans{
90 Req: reqWorker,
91 Resp: respWorker,
92 Ready: make(chan wire.Ready, 1),
93 Stop: t.stop,
94 }))
95 }
96
97 workers := cfg.ConnWorkers
98 if cfg.MaxConnections > 0 && cfg.MaxConnections < workers {
99 workers = cfg.MaxConnections
100 }
101 acceptorCh := iodomain.AcceptorChans{Stop: t.stop}
102 for i := 0; i < workers; i++ {
103 assignAcceptor, assignReader := tinyspawn.NewBufferedDuplex[wire.Assign](1)
104 inReader, inNet := tinyspawn.NewBufferedDuplex[wire.Inbound](64)
105 outNet, outWriter := tinyspawn.NewBufferedDuplex[wire.Outbound](64)
106 ctlReader, ctlWriter := tinyspawn.NewBufferedDuplex[wire.WriterCtl](32)
107 ackReader, ackWriter := tinyspawn.NewBufferedDuplex[uint8](2)
108
109 netCh.Workers = append(netCh.Workers, netdomain.WorkerLink{In: inNet, Out: outNet})
110 acceptorCh.Workers = append(acceptorCh.Workers, assignAcceptor)
111
112 t.add("reader", tinyspawn.Spawn(iodomain.Reader, iodomain.ReaderData{Index: i, Cfg: cfg}, iodomain.ReaderChans{
113 Assign: assignReader,
114 In: inReader,
115 Ctl: ctlReader,
116 Ack: ackReader,
117 Stop: t.stop,
118 }))
119 t.add("writer", tinyspawn.Spawn(iodomain.Writer, iodomain.WriterData{Index: i}, iodomain.WriterChans{
120 Ctl: ctlWriter,
121 Out: outWriter,
122 Ack: ackWriter,
123 Stop: t.stop,
124 }))
125 }
126
127 t.add("net", tinyspawn.Spawn(netdomain.Run, netdomain.Data{Cfg: cfg, RelaySk: sk}, netCh))
128
129 acceptorReady := make(chan wire.Ready, 1)
130 acceptorCh.Ready = acceptorReady
131 t.add("acceptor", tinyspawn.Spawn(iodomain.Acceptor, iodomain.AcceptorData{Addr: cfg.Addr(), Cfg: cfg}, acceptorCh))
132 select {
133 case r := <-acceptorReady:
134 if !r.OK {
135 t.Close()
136 return nil, errors.New("acceptor: " + r.Err)
137 }
138 if r.Addr != "" {
139 t.addr = r.Addr
140 }
141 case <-time.After(10 * time.Second):
142 t.Close()
143 return nil, errors.New("acceptor: startup timed out")
144 }
145 return t, nil
146 }
147
148 func (t *Tree) add(_ string, done <-chan struct{}) { t.done = append(t.done, done) }
149
150 func waitReady(ch <-chan wire.Ready) error {
151 return waitReadyFor(ch, 10*time.Second)
152 }
153
154 func waitReadyFor(ch <-chan wire.Ready, limit time.Duration) error {
155 select {
156 case r := <-ch:
157 if !r.OK {
158 return errors.New(r.Err)
159 }
160 return nil
161 case <-time.After(limit):
162 return errors.New("startup timed out")
163 }
164 }
165
166 // Addr returns the listening address.
167 func (t *Tree) Addr() string { return t.addr }
168
169 // Config returns the immutable configuration used to start the tree.
170 func (t *Tree) Config() config.Config { return t.cfg }
171
172 // Exited reports whether any domain has returned.
173 func (t *Tree) Exited() bool {
174 for _, d := range t.done {
175 select {
176 case <-d:
177 return true
178 default:
179 }
180 }
181 return false
182 }
183
184 // Close cancels the tree, closes the listening socket and waits for every
185 // domain to return.
186 func (t *Tree) Close() {
187 t.once.Do(func() { t.stop.Close() })
188 deadline := time.Now().Add(5 * time.Second)
189 stopped := 0
190 for _, d := range t.done {
191 left := time.Until(deadline)
192 if left <= 0 {
193 break
194 }
195 select {
196 case <-d:
197 stopped++
198 case <-time.After(left):
199 }
200 }
201
202 }
203