| 1 | package coreapi |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | |
| 9 | blockservice "github.com/ipfs/boxo/blockservice" |
| 10 | blockstore "github.com/ipfs/boxo/blockstore" |
| 11 | offline "github.com/ipfs/boxo/exchange/offline" |
| 12 | dag "github.com/ipfs/boxo/ipld/merkledag" |
| 13 | "github.com/ipfs/boxo/path" |
| 14 | cid "github.com/ipfs/go-cid" |
| 15 | cidutil "github.com/ipfs/go-cidutil" |
| 16 | coreiface "github.com/ipfs/kubo/core/coreiface" |
| 17 | caopts "github.com/ipfs/kubo/core/coreiface/options" |
| 18 | "github.com/ipfs/kubo/core/node" |
| 19 | "github.com/ipfs/kubo/tracing" |
| 20 | peer "github.com/libp2p/go-libp2p/core/peer" |
| 21 | mh "github.com/multiformats/go-multihash" |
| 22 | "go.opentelemetry.io/otel/attribute" |
| 23 | "go.opentelemetry.io/otel/trace" |
| 24 | ) |
| 25 | |
| 26 | type RoutingAPI CoreAPI |
| 27 | |
| 28 | func (api *RoutingAPI) Get(ctx context.Context, key string) ([]byte, error) { |
| 29 | if !api.nd.IsOnline { |
| 30 | return nil, coreiface.ErrOffline |
| 31 | } |
| 32 | |
| 33 | dhtKey, err := normalizeKey(key) |
| 34 | if err != nil { |
| 35 | return nil, err |
| 36 | } |
| 37 | |
| 38 | return api.routing.GetValue(ctx, dhtKey) |
| 39 | } |
| 40 | |
| 41 | func (api *RoutingAPI) Put(ctx context.Context, key string, value []byte, opts ...caopts.RoutingPutOption) error { |
| 42 | options, err := caopts.RoutingPutOptions(opts...) |
| 43 | if err != nil { |
| 44 | return err |
| 45 | } |
| 46 | |
| 47 | err = api.checkOnline(options.AllowOffline) |
| 48 | if err != nil { |
| 49 | return err |
| 50 | } |
| 51 | |
| 52 | dhtKey, err := normalizeKey(key) |
| 53 | if err != nil { |
| 54 | return err |
| 55 | } |
| 56 | |
| 57 | return api.routing.PutValue(ctx, dhtKey, value) |
| 58 | } |
| 59 | |
| 60 | func normalizeKey(s string) (string, error) { |
| 61 | parts := strings.Split(s, "/") |
| 62 | if len(parts) != 3 || |
| 63 | parts[0] != "" || |
| 64 | !(parts[1] == "ipns" || parts[1] == "pk") { |
| 65 | return "", errors.New("invalid key") |
| 66 | } |
| 67 | |
| 68 | k, err := peer.Decode(parts[2]) |
| 69 | if err != nil { |
| 70 | return "", err |
| 71 | } |
| 72 | return strings.Join(append(parts[:2], string(k)), "/"), nil |
| 73 | } |
| 74 | |
| 75 | func (api *RoutingAPI) FindPeer(ctx context.Context, p peer.ID) (peer.AddrInfo, error) { |
| 76 | ctx, span := tracing.Span(ctx, "CoreAPI.DhtAPI", "FindPeer", trace.WithAttributes(attribute.String("peer", p.String()))) |
| 77 | defer span.End() |
| 78 | err := api.checkOnline(false) |
| 79 | if err != nil { |
| 80 | return peer.AddrInfo{}, err |
| 81 | } |
| 82 | |
| 83 | pi, err := api.routing.FindPeer(ctx, peer.ID(p)) |
| 84 | if err != nil { |
| 85 | return peer.AddrInfo{}, err |
| 86 | } |
| 87 | |
| 88 | return pi, nil |
| 89 | } |
| 90 | |
| 91 | func (api *RoutingAPI) FindProviders(ctx context.Context, p path.Path, opts ...caopts.RoutingFindProvidersOption) (<-chan peer.AddrInfo, error) { |
| 92 | ctx, span := tracing.Span(ctx, "CoreAPI.DhtAPI", "FindProviders", trace.WithAttributes(attribute.String("path", p.String()))) |
| 93 | defer span.End() |
| 94 | |
| 95 | settings, err := caopts.RoutingFindProvidersOptions(opts...) |
| 96 | if err != nil { |
| 97 | return nil, err |
| 98 | } |
| 99 | span.SetAttributes(attribute.Int("numproviders", settings.NumProviders)) |
| 100 | |
| 101 | err = api.checkOnline(false) |
| 102 | if err != nil { |
| 103 | return nil, err |
| 104 | } |
| 105 | |
| 106 | rp, _, err := api.core().ResolvePath(ctx, p) |
| 107 | if err != nil { |
| 108 | return nil, err |
| 109 | } |
| 110 | |
| 111 | numProviders := settings.NumProviders |
| 112 | if numProviders < 1 { |
| 113 | return nil, errors.New("number of providers must be greater than 0") |
| 114 | } |
| 115 | |
| 116 | pchan := api.routing.FindProvidersAsync(ctx, rp.RootCid(), numProviders) |
| 117 | return pchan, nil |
| 118 | } |
| 119 | |
| 120 | func (api *RoutingAPI) Provide(ctx context.Context, path path.Path, opts ...caopts.RoutingProvideOption) error { |
| 121 | ctx, span := tracing.Span(ctx, "CoreAPI.DhtAPI", "Provide", trace.WithAttributes(attribute.String("path", path.String()))) |
| 122 | defer span.End() |
| 123 | |
| 124 | settings, err := caopts.RoutingProvideOptions(opts...) |
| 125 | if err != nil { |
| 126 | return err |
| 127 | } |
| 128 | span.SetAttributes(attribute.Bool("recursive", settings.Recursive)) |
| 129 | |
| 130 | err = api.checkOnline(false) |
| 131 | if err != nil { |
| 132 | return err |
| 133 | } |
| 134 | |
| 135 | rp, _, err := api.core().ResolvePath(ctx, path) |
| 136 | if err != nil { |
| 137 | return err |
| 138 | } |
| 139 | |
| 140 | c := rp.RootCid() |
| 141 | |
| 142 | has, err := api.blockstore.Has(ctx, c) |
| 143 | if err != nil { |
| 144 | return err |
| 145 | } |
| 146 | |
| 147 | if !has { |
| 148 | return fmt.Errorf("block %s not found locally, cannot provide", c) |
| 149 | } |
| 150 | |
| 151 | if settings.Recursive { |
| 152 | err = provideKeysRec(ctx, api.provider, api.blockstore, []cid.Cid{c}) |
| 153 | } else { |
| 154 | err = api.provider.StartProviding(false, c.Hash()) |
| 155 | } |
| 156 | if err != nil { |
| 157 | return err |
| 158 | } |
| 159 | |
| 160 | return nil |
| 161 | } |
| 162 | |
| 163 | func provideKeysRec(ctx context.Context, prov node.DHTProvider, bs blockstore.Blockstore, cids []cid.Cid) error { |
| 164 | provided := cidutil.NewStreamingSet() |
| 165 | |
| 166 | // Error channel with buffer size 1 to avoid blocking the goroutine |
| 167 | errCh := make(chan error, 1) |
| 168 | go func() { |
| 169 | // Always close provided.New to signal completion |
| 170 | defer close(provided.New) |
| 171 | // Also close error channel to distinguish between "no error" and "pending error" |
| 172 | defer close(errCh) |
| 173 | |
| 174 | dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs))) |
| 175 | for _, c := range cids { |
| 176 | if err := dag.Walk(ctx, dag.GetLinksDirect(dserv), c, provided.Visitor(ctx)); err != nil { |
| 177 | // Send error to channel. If context is cancelled while trying to send, |
| 178 | // exit immediately as the main loop will return ctx.Err() |
| 179 | select { |
| 180 | case errCh <- err: |
| 181 | // Error sent successfully, exit goroutine |
| 182 | case <-ctx.Done(): |
| 183 | // Context cancelled, exit without sending error |
| 184 | return |
| 185 | } |
| 186 | return |
| 187 | } |
| 188 | } |
| 189 | // All CIDs walked successfully, goroutine will exit and channels will close |
| 190 | }() |
| 191 | |
| 192 | keys := make([]mh.Multihash, 0) |
| 193 | for { |
| 194 | select { |
| 195 | case <-ctx.Done(): |
| 196 | // Context cancelled, return immediately |
| 197 | return ctx.Err() |
| 198 | case err := <-errCh: |
| 199 | // Received error from DAG walk, return it |
| 200 | return err |
| 201 | case c, ok := <-provided.New: |
| 202 | if !ok { |
| 203 | // Channel closed means goroutine finished. |
| 204 | // CRITICAL: Check for any error that was sent just before channel closure. |
| 205 | // This handles the race where error is sent to errCh but main loop |
| 206 | // sees provided.New close first. |
| 207 | select { |
| 208 | case err := <-errCh: |
| 209 | if err != nil { |
| 210 | return err |
| 211 | } |
| 212 | // errCh closed with nil, meaning success |
| 213 | default: |
| 214 | // No pending error in errCh |
| 215 | } |
| 216 | // All CIDs successfully processed, start providing |
| 217 | return prov.StartProviding(true, keys...) |
| 218 | } |
| 219 | // Accumulate the CID for providing |
| 220 | keys = append(keys, c.Hash()) |
| 221 | } |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | func (api *RoutingAPI) core() coreiface.CoreAPI { |
| 226 | return (*CoreAPI)(api) |
| 227 | } |