diff --git a/.github/workflows/trivy.yaml b/.github/workflows/trivy.yaml index 0491292..83fb1df 100644 --- a/.github/workflows/trivy.yaml +++ b/.github/workflows/trivy.yaml @@ -29,8 +29,6 @@ jobs: - name: Run Trivy vulnerability scanner uses: aquasecurity/trivy-action@57a97c7e7821a5776cebc9bb87c984fa69cba8f1 # 0.35.0 - env: - TRIVY_DB_REPOSITORY: public.ecr.aws/aquasecurity/trivy-db:2 with: image-ref: "${{ env.IMAGE_NAME }}:${{ github.sha }}" format: "table" diff --git a/README.md b/README.md index ca775d8..c844104 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ owner is offline. A wallet registers a watch on its address; every spend is co-signed by the [emulator](https://github.com/arkade-os/emulator) only if it satisfies the template's covenant. -The default templates in [`templates/`](templates) (renewal, boarding, VHTLC claim and refund, and noninteractiveclaim) are registered and trusted at startup. +The default templates in [`templates/`](templates) (renewal, boarding, VHTLC claim and refund, noninteractiveclaim, and offline_receive) are registered and trusted at startup. ## Run @@ -79,5 +79,73 @@ belongs to one pool, and each pool has one output without an `amount`. ## Stream delegations -A template with a `trigger` is not registered by clients: the operator registers it as a stream delegation (admin `RegisterStreamDelegation`; a default with a trigger starts by itself when an encryption key is configured). The daemon reads arkd's transaction stream and, for every transaction whose extension holds the trigger's packet, binds the trigger's variables, checks the template locks the output, and spawns a watch of the template that expires after `expires_after_seconds`. - +Offline receive uses `templates/offline_receive.json` as an operator stream delegation. +The receiver never registers a watch or sends its preimage to a delegatee API. Before +requesting an exact-output quote, it seals the preimage, recipient and minimum amount +to the daemon's advertised encryption public key. The operator can decrypt the preimage +and must be trusted with it. The funding transaction delivers +that ciphertext in extension packet type 4. The solver adds its payout script in TLV 2; +the daemon checks the full funding taptree and output script against the resulting +covenant claim leaf, and authenticates its recipient against the envelope before claiming. + +The packet body starts with TLV 127 containing ASCII `delegatee-secrets/v1`, followed +by TLV 1 (ciphertext) and TLV 3 (compressed encryption public key). Each record is a +one-byte tag, two-byte big-endian length and value. The ciphertext uses the existing +ECIES wire format and `covclaimd/preimage/v1` domain, encrypting JSON: +`{"format":"delegatee-secrets/v1","template_id":"","values":{...}}`. +The values are `preimage` and `receiver_program` (32-byte lowercase hex) and +`minimum_value` (canonical ScriptNum hex). Secret declarations select fields with +`field`. The recipient captured from TLV 2 must equal the authenticated receiver before +a claim can disclose the preimage. Raw ECIES plaintext remains supported by +templates without `field`, including the unchanged `noninteractiveclaim` stream. + +The authenticated minimum must be positive, at most 2.1e15 satoshis, and satisfied by +the funded input before a claim can disclose the preimage. The output preserves the +input's entire value and assets and pays the authenticated recipient. This envelope +binds an intent before its quote, not a specific later refund deadline or lockup tree. +There is no encrypted claim deadline; ordinary spawned watches expire after 24 hours. + +The receiver can go offline after preparing its invoice. Once the daemon receives the +funding packet, the spawned watch persists and resumes retries after a daemon restart. +The filtered indexer subscription does not replay history: funding first seen while the daemon +is unavailable cannot currently be discovered through this packet-only flow. The +indexer recovers known watches, not entirely unseen funding packets. + +A template with a `trigger` is not registered by clients: the operator registers it as a stream delegation (admin `RegisterStreamDelegation`; a default with a trigger starts by itself when an encryption key is configured). The daemon opens an indexer subscription with CEL filters derived from the active triggers. For offline receive these select packet type 4, the version marker and one of its retained encryption keys. It binds matching packets, checks the template locks the output, and spawns a watch that expires after `expires_after_seconds`. Full local trigger and covenant checks remain authoritative: the CEL filter only reduces traffic. This requires indexer initial-filter support; subscription errors are reported and retried without switching to the global transaction stream. + + +## Capacity + +Discovery filter changes update the existing subscription atomically while its +events continue to be processed. Failed or ambiguous updates are reconciled +without closing the working stream. + +Known offchain direct watches use script subscriptions, split into at most 1,000 +scripts per subscription. Existing shards retain their scripts as registrations +change. New and spent coin events invalidate only the affected scripts; repeated +events coalesce. The next scan queries those scripts in chunks of 100 with at +most four requests in flight. Startup, connection gaps and periodic reconciliation +refresh all active scripts. Frequent events cannot postpone the periodic refresh. +Upstream throttle backoff takes precedence over event wakes and reconciliation +deadlines; failed scans retain their dirty scripts for the next permitted retry. + +Packet discovery needs a recipient filter because a new payment's script is not +known beforehand. It uses one separate filtered subscription, with a one-event +application buffer and synchronous registration. This bounds memory, but does not +provide durable event replay or guaranteed admission during overload, a database +outage or a full registration cap. Watch rows already persisted are recoverable. + +The default caps are 50,000 public/stream registrations and 10,000 spawned watches. +A successful receive watch remains active until its 24-hour expiry so it can handle +another funding to that script. Consequently 10,000 distinct receives can fill the +child cap during a day even when claims settle quickly. Raising the cap requires +capacity testing; it does not increase processing throughput. + +All active registrations are still loaded and examined each scan, and direct claims +run serially within their lane. Registration inserts serialize and count active +rows. Batch planning uses sorted passes rather than repeatedly copying/scanning +remaining entries. Unit scale tests cover 50,000 scripts, bounded subscriptions, +coalesced refreshes, races and failed/reconnected streams; these use mock storage +and indexer responses and do not establish production throughput. Measure database +latency, scan duration, claim service time, oldest pending receive and cap usage +under the deployment's actual traffic before increasing limits. diff --git a/go.mod b/go.mod index a67e9ad..0d9d685 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ replace github.com/arkade-os/delegatee/api-spec => ./api-spec replace github.com/arkade-os/delegatee/pkg/template => ./pkg/template require ( + github.com/arkade-os/arkd/api-spec v0.0.0-20260901090427-f863e4847193 github.com/arkade-os/arkd/pkg/ark-lib v0.8.1-0.20260901090427-f863e4847193 github.com/arkade-os/arkd/pkg/client-lib v0.0.0-20260901090427-f863e4847193 github.com/arkade-os/arkd/pkg/client-wallet v0.0.0-20260901090427-f863e4847193 @@ -40,7 +41,6 @@ require ( cel.dev/expr v0.25.2 // indirect github.com/aead/siphash v1.0.1 // indirect github.com/antlr4-go/antlr/v4 v4.13.1 // indirect - github.com/arkade-os/arkd/api-spec v0.0.0-20260901090427-f863e4847193 // indirect github.com/arkade-os/arkd/pkg/errors v0.0.0-20260901090427-f863e4847193 // indirect github.com/arkade-os/emulator/api-spec v0.1.0 // indirect github.com/beorn7/perks v1.0.1 // indirect diff --git a/internal/config/config.go b/internal/config/config.go index 521b6db..c286dcd 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -14,11 +14,11 @@ import ( clientlib "github.com/arkade-os/arkd/pkg/client-lib" "github.com/arkade-os/arkd/pkg/client-lib/client" - "github.com/arkade-os/arkd/pkg/client-lib/indexer" "github.com/arkade-os/delegatee/internal/core/application" "github.com/arkade-os/delegatee/internal/core/domain" "github.com/arkade-os/delegatee/internal/infrastructure/db/postgres" "github.com/arkade-os/delegatee/internal/infrastructure/explorer" + "github.com/arkade-os/delegatee/internal/infrastructure/indexer" "github.com/arkade-os/delegatee/templates" emulatorclient "github.com/arkade-os/emulator/pkg/client" "github.com/btcsuite/btcd/btcec/v2" diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5dd92a3..9d36905 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -255,12 +255,13 @@ func TestRegisterDefaults(t *testing.T) { renewal, boarding := "d901cb8554a77524fb55ebfcde8e1c468871de45163abf34e988c47f1fa36896", "02b4e8bdfb52d7450ab75510cfaff36da5b88323291325e38358df10684a8344" claim, refund := "c6e10a94e6b27b1c29dc3c9eae95d1f01ef9e4142f882f9ce9888a416d7ae01d", "978df861b2ae33f1373fd9ead844b074b3575d2341e98f1d50114179a9ed7d6d" noninteractiveclaim := "4493579d4cdb14a5809b55c55b7d228a4d8f7fc479c1424b286d4d5234382eee" + offlineReceive := "fe39de17a0f608b710d9290b7533ec48204493a3e7cf726e03098b996eea38a2" on := &Config{DefaultTemplates: true} svc := newDefaultsService() require.NoError(t, on.registerDefaults(t.Context(), svc)) require.Len(t, svc.artifacts, 1) - require.Equal(t, map[string]bool{renewal: true, boarding: true, claim: true, refund: true, noninteractiveclaim: true}, svc.trusted()) - require.Equal(t, map[string]bool{noninteractiveclaim: true}, svc.streams, "only the template with a trigger streams") + require.Equal(t, map[string]bool{renewal: true, boarding: true, claim: true, refund: true, noninteractiveclaim: true, offlineReceive: true}, svc.trusted()) + require.Equal(t, map[string]bool{noninteractiveclaim: true, offlineReceive: true}, svc.streams, "templates with triggers stream") noKey := newDefaultsService() noKey.streamErr = application.ErrSecretsRequired @@ -270,7 +271,7 @@ func TestRegisterDefaults(t *testing.T) { trusts := svc.trusts require.NoError(t, on.registerDefaults(t.Context(), svc)) require.Equal(t, trusts, svc.trusts, "a restart changes nothing") - require.Equal(t, map[string]bool{renewal: true, boarding: false, claim: true, refund: true, noninteractiveclaim: true}, svc.trusted(), "an untrusted default stays untrusted") + require.Equal(t, map[string]bool{renewal: true, boarding: false, claim: true, refund: true, noninteractiveclaim: true, offlineReceive: true}, svc.trusted(), "an untrusted default stays untrusted") off := newDefaultsService() require.NoError(t, (&Config{}).registerDefaults(t.Context(), off)) diff --git a/internal/core/application/collection.go b/internal/core/application/collection.go index 999102a..c2bb48f 100644 --- a/internal/core/application/collection.go +++ b/internal/core/application/collection.go @@ -60,19 +60,29 @@ func (s *service) planBatches(entries []planEntry, now time.Time) *plan { slices.SortStableFunc(vtxos, func(a, b planEntry) int { return cmp.Or(a.deadline.Compare(b.deadline), a.due.Compare(b.due)) }) - // deadline >= due: the entry with the earliest deadline is always in its batch, so this ends - for len(vtxos) > 0 { - at := vtxos[0].deadline - b := PlannedBatch{At: at, ForcedBy: vtxos[0].outpoints[0], ForcedByDelegation: vtxos[0].delegation} - var rest []planEntry - for _, e := range vtxos { + dueOrder := make([]int, len(vtxos)) + for i := range dueOrder { + dueOrder[i] = i + } + slices.SortFunc(dueOrder, func(a, b int) int { return vtxos[a].due.Compare(vtxos[b].due) }) + taken := make([]bool, len(vtxos)) + nextDue := 0 + for i, first := range vtxos { + if taken[i] { + continue + } + at := first.deadline + b := PlannedBatch{At: at, ForcedBy: first.outpoints[0], ForcedByDelegation: first.delegation} + for nextDue < len(dueOrder) { + index := dueOrder[nextDue] + e := vtxos[index] if e.due.After(at) { - rest = append(rest, e) - continue + break } p.join(&b, e, at) + taken[index] = true + nextDue++ } - vtxos = rest p.batches = append(p.batches, b) } for _, e := range deposits { diff --git a/internal/core/application/collection_test.go b/internal/core/application/collection_test.go index 775b583..85d3811 100644 --- a/internal/core/application/collection_test.go +++ b/internal/core/application/collection_test.go @@ -1,7 +1,11 @@ package application import ( + "cmp" "errors" + "fmt" + "math/rand/v2" + "slices" "testing" "testing/synctest" "time" @@ -234,3 +238,54 @@ func TestReserveMustCoverARetry(t *testing.T) { []*btcec.PrivateKey{key}, time.Hour, time.Minute, 2*time.Minute, 0, env.svc.limits) require.NoError(t, err) } + +func BenchmarkPlanScatteredDeadlines(b *testing.B) { + now := time.Now() + entries := make([]planEntry, 5000) + for n := range entries { + at := now.Add(time.Duration(n+1) * time.Second) + entries[n] = planEntry{outpoints: []string{fmt.Sprintf("coin-%d", n)}, delegation: int64(n), amount: 1000, due: at, deadline: at} + } + svc := &service{} + b.ReportAllocs() + b.ResetTimer() + for range b.N { + plan := svc.planBatches(entries, now) + if len(plan.batches) != len(entries) { + b.Fatal("scattered deadlines lost") + } + } +} + +func TestPlanBatchSweepPreservesAssignments(t *testing.T) { + rng := rand.New(rand.NewPCG(1, 2)) + now := time.Now() + svc := &service{boardingMaxWait: 10 * time.Minute} + for range 100 { + entries := make([]planEntry, 100) + for i := range entries { + due := now.Add(time.Duration(rng.IntN(100)-20) * time.Minute) + entries[i] = planEntry{outpoints: []string{fmt.Sprintf("coin-%d", i)}, delegation: int64(i), amount: uint64(i + 1), due: due, deadline: due.Add(time.Duration(rng.IntN(20)) * time.Minute)} + } + remaining := slices.Clone(entries) + slices.SortStableFunc(remaining, func(a, b planEntry) int { return cmp.Or(a.deadline.Compare(b.deadline), a.due.Compare(b.due)) }) + expected := &plan{batchAt: map[string]time.Time{}} + for len(remaining) > 0 { + first := remaining[0] + at := first.deadline + batch := PlannedBatch{At: at, ForcedBy: first.outpoints[0], ForcedByDelegation: first.delegation} + var later []planEntry + for _, e := range remaining { + if e.due.After(at) { + later = append(later, e) + } else { + expected.join(&batch, e, at) + } + } + batch.At = maxTime(batch.At, now) + expected.batches = append(expected.batches, batch) + remaining = later + } + require.Equal(t, expected, svc.planBatches(entries, now)) + } +} diff --git a/internal/core/application/discovery.go b/internal/core/application/discovery.go index e850e1d..fd51160 100644 --- a/internal/core/application/discovery.go +++ b/internal/core/application/discovery.go @@ -276,7 +276,7 @@ func (s *service) discover(ctx context.Context, delegations []domain.Delegation, unwatched := map[int64]string{} s.found = map[int64][]scannedCoin{} scripts := s.watchAll(ctx, delegations, active, unwatched) - byScript, err := s.spendableVtxos(ctx, scripts) + byScript, err := s.scanVtxos(ctx, scripts) if err != nil { return nil, nil, nil, err } diff --git a/internal/core/application/helpers_test.go b/internal/core/application/helpers_test.go index a653d32..b228a94 100644 --- a/internal/core/application/helpers_test.go +++ b/internal/core/application/helpers_test.go @@ -580,8 +580,6 @@ type fakeArk struct { // pending answers GetPendingTx for a proof signed for its coins pending []clientlib.AcceptedOffchainTx pendingProofs int - txs chan clientlib.TransactionEvent // GetTransactionsStream; nil refuses it - txStreams int } func (a *fakeArk) GetPendingTx(_ context.Context, proof, message string) ([]clientlib.AcceptedOffchainTx, error) { @@ -697,30 +695,6 @@ func (a *fakeArk) GetEventStream(context.Context, []string) (<-chan clientlib.Ba return ch, func() {}, nil } -func (a *fakeArk) GetTransactionsStream(context.Context) (<-chan clientlib.TransactionEvent, func(), error) { - a.mu.Lock() - defer a.mu.Unlock() - if a.txs == nil { - return nil, nil, errors.New("no transaction stream") - } - a.txStreams++ - return a.txs, func() {}, nil -} - -func (a *fakeArk) setTxs(txs chan clientlib.TransactionEvent) { - a.mu.Lock() - defer a.mu.Unlock() - a.txs = txs -} - -func (a *fakeArk) opened(n int) func() bool { - return func() bool { - a.mu.Lock() - defer a.mu.Unlock() - return a.txStreams == n - } -} - func (a *fakeArk) SubmitTx(_ context.Context, ark string, checkpoints []string) (string, string, []string, error) { return a.submitTx(ark, checkpoints) } @@ -739,6 +713,12 @@ func (a *fakeArk) FinalizeTx(_ context.Context, _ string, checkpoints []string) } type fakeIndexer struct { + txUpdates int + updateTransactions func() error + transactionEvents chan clientlib.TransactionEvent + txStreams int + filters [][]string + subScripts map[string]map[string]bool clientlib.Indexer mu sync.Mutex vtxos []clientlib.Vtxo @@ -759,6 +739,47 @@ type fakeIndexer struct { opened [][]string // scripts each NewSubscription received } +func (a *fakeIndexer) NewTransactionSubscription(_ context.Context, expressions []string) (string, <-chan clientlib.TransactionEvent, func(), error) { + a.mu.Lock() + defer a.mu.Unlock() + if a.transactionEvents == nil { + return "", nil, nil, errors.New("no transaction stream") + } + a.txStreams++ + a.filters = append(a.filters, slices.Clone(expressions)) + return "transactions", a.transactionEvents, func() {}, nil +} + +func (a *fakeIndexer) UpdateTransactionSubscription(_ context.Context, id string, expressions []string) error { + a.mu.Lock() + a.txUpdates++ + update := a.updateTransactions + a.mu.Unlock() + if update != nil { + if err := update(); err != nil { + return err + } + } + a.mu.Lock() + defer a.mu.Unlock() + a.filters = append(a.filters, slices.Clone(expressions)) + return nil +} + +func (a *fakeIndexer) setTxs(transactionEvents chan clientlib.TransactionEvent) { + a.mu.Lock() + defer a.mu.Unlock() + a.transactionEvents = transactionEvents +} + +func (a *fakeIndexer) transactionsOpened(n int) func() bool { + return func() bool { + a.mu.Lock() + defer a.mu.Unlock() + return a.txStreams == n + } +} + // GetVtxos serves everything to a scan's first request by scripts. func (i *fakeIndexer) GetVtxos(_ context.Context, opts ...clientlib.GetVtxosOption) (*clientlib.VtxosResponse, error) { i.mu.Lock() @@ -792,39 +813,64 @@ func (i *fakeIndexer) NewSubscription(_ context.Context, scripts []string) (stri if i.subErr != nil { return "", nil, nil, i.subErr } + if len(scripts) > maxSubscriptionScripts { + return "", nil, nil, errors.New("too many scripts") + } i.subscriptions++ + id := fmt.Sprintf("sub-%d", i.subscriptions) i.opened = append(i.opened, slices.Clone(scripts)) - i.subscribed = map[string]bool{} + if i.subScripts == nil { + i.subScripts = map[string]map[string]bool{} + } + i.subScripts[id] = map[string]bool{} for _, script := range scripts { - i.subscribed[script] = true + i.subScripts[id][script] = true } + i.updateSubscribed() events := make(chan clientlib.ScriptEvent, 8) - i.events = events - i.endEvents = sync.OnceFunc(func() { close(events) }) - stop := i.endEvents - return fmt.Sprintf("sub-%d", i.subscriptions), events, func() { - stop() + stop := sync.OnceFunc(func() { i.mu.Lock() defer i.mu.Unlock() - i.subscribed = nil - }, nil + close(events) + delete(i.subScripts, id) + i.updateSubscribed() + }) + i.events, i.endEvents = events, stop + return id, events, stop, nil } -func (i *fakeIndexer) UpdateSubscription(_ context.Context, _ string, add, remove []string) error { +func (i *fakeIndexer) UpdateSubscription(_ context.Context, id string, add, remove []string) error { i.mu.Lock() defer i.mu.Unlock() if i.updateErr != nil { return i.updateErr } + scripts, ok := i.subScripts[id] + if !ok { + return errors.New("subscription missing") + } + next := maps.Clone(scripts) for _, script := range add { - i.subscribed[script] = true + next[script] = true } for _, script := range remove { - delete(i.subscribed, script) + delete(next, script) } + if len(next) > maxSubscriptionScripts { + return errors.New("too many scripts") + } + i.subScripts[id] = next + i.updateSubscribed() return nil } +func (i *fakeIndexer) updateSubscribed() { + i.subscribed = map[string]bool{} + for _, scripts := range i.subScripts { + maps.Copy(i.subscribed, scripts) + } +} + func (i *fakeIndexer) subscribedScripts() []string { i.mu.Lock() defer i.mu.Unlock() @@ -834,8 +880,9 @@ func (i *fakeIndexer) subscribedScripts() []string { // closeSubscription ends the stream as arkd going away for good does. func (i *fakeIndexer) closeSubscription() { i.mu.Lock() - defer i.mu.Unlock() - i.endEvents() + stop := i.endEvents + i.mu.Unlock() + stop() } func (i *fakeIndexer) GetVirtualTxs(_ context.Context, txids []string, _ ...clientlib.PageOption) (*clientlib.VirtualTxsResponse, error) { @@ -1496,7 +1543,7 @@ func (e *testEnv) openSubscription(t *testing.T) { t.Helper() e.indexer.serve() e.scan(t) - require.NotNil(t, e.svc.sub) + require.NotEmpty(t, e.svc.subs) for len(e.svc.wake) > 0 { <-e.svc.wake } diff --git a/internal/core/application/lanes.go b/internal/core/application/lanes.go index 3dbb3c6..5da96b8 100644 --- a/internal/core/application/lanes.go +++ b/internal/core/application/lanes.go @@ -68,6 +68,9 @@ func (s *service) dispatch(l *lane, inputs []renewalInput) { s.renewAndRecord(context.Background(), inputs) s.setInFlight(inputs, false) l.busy.Store(0) + for _, in := range inputs { + s.markScriptsDirty(in.watched.delegation.Slots[in.coin.Slot].Script) + } if l.waiting.Swap(false) { s.wakeScan() } diff --git a/internal/core/application/offline_packet_test.go b/internal/core/application/offline_packet_test.go new file mode 100644 index 0000000..6b5ee82 --- /dev/null +++ b/internal/core/application/offline_packet_test.go @@ -0,0 +1,18 @@ +package application + +import ( + "testing" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/stretchr/testify/require" +) + +func TestOfflineReceivePacketStartsWithoutClientRegistration(t *testing.T) { + e := newTestEnv(t) + key, _ := hexKey(t) + e.svc.encryptionKeys = []*btcec.PrivateKey{key} + id := e.trust(t, e.fixture(t, "offline_receive.json")) + stream, err := e.svc.RegisterStreamDelegation(t.Context(), id, nil) + require.NoError(t, err) + require.True(t, stream.IsStream()) +} diff --git a/internal/core/application/offline_receive_test.go b/internal/core/application/offline_receive_test.go new file mode 100644 index 0000000..e86f81e --- /dev/null +++ b/internal/core/application/offline_receive_test.go @@ -0,0 +1,281 @@ +package application + +import ( + "bytes" + "context" + "encoding/binary" + "encoding/hex" + "encoding/json" + "maps" + "testing" + "time" + + "github.com/arkade-os/arkd/pkg/ark-lib/extension" + "github.com/arkade-os/arkd/pkg/ark-lib/txutils" + clientlib "github.com/arkade-os/arkd/pkg/client-lib" + "github.com/arkade-os/delegatee/internal/core/domain" + "github.com/arkade-os/delegatee/pkg/template" + "github.com/arkade-os/delegatee/pkg/template/ecies" + "github.com/arkade-os/delegatee/pkg/template/packets" + "github.com/btcsuite/btcd/btcec/v2" + "github.com/btcsuite/btcd/psbt/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func offlineStream(t *testing.T) (*testEnv, *btcec.PrivateKey, *domain.Delegation) { + t.Helper() + e, _ := serverEnv(t) + key, _ := hexKey(t) + e.svc.encryptionKeys = []*btcec.PrivateKey{key} + d, err := e.svc.RegisterStreamDelegation(t.Context(), e.trust(t, e.fixture(t, "offline_receive.json")), nil) + require.NoError(t, err) + e.scan(t) + return e, key, d +} + +func offlinePacket(ciphertext, script, key []byte) []byte { + marker := []byte("delegatee-secrets/v1") + body := binary.BigEndian.AppendUint16([]byte{127}, uint16(len(marker))) + body = append(body, marker...) + body = append(body, claimPacket(ciphertext, nil, key)...) + return append(body, claimPacket(nil, script, nil)...) +} + +func offlineFunding(t *testing.T, e *testEnv, key *btcec.PrivateKey, id string, alter func(map[string]any)) (*psbt.Packet, map[string]string) { + t.Helper() + ptx, legacy := funding(t, e, key.PubKey()) + ct, err := hex.DecodeString(legacy["ciphertext"]) + require.NoError(t, err) + preimage, err := ecies.Decrypt(key, ct) + require.NoError(t, err) + envelope := map[string]any{"format": "delegatee-secrets/v1", "template_id": id, "values": map[string]string{ + "preimage": hex.EncodeToString(preimage), "minimum_value": hex.EncodeToString(scriptNum(10_000)), "receiver_program": legacy["receiver_program"], + }} + if alter != nil { + alter(envelope) + } + raw, err := json.Marshal(envelope) + require.NoError(t, err) + ct, err = ecies.Encrypt(key.PubKey(), raw) + require.NoError(t, err) + program, err := hex.DecodeString(legacy["receiver_program"]) + require.NoError(t, err) + setOfflinePacket(t, ptx, offlinePacket(ct, enforcePayTo(t, program), key.PubKey().SerializeCompressed())) + return ptx, legacy +} + +func setOfflinePacket(t *testing.T, ptx *psbt.Packet, body []byte) { + t.Helper() + out, err := extension.Extension{packets.Raw(4, body)}.TxOut() + require.NoError(t, err) + ptx.UnsignedTx.TxOut[1] = out +} + +func TestOfflineReceivePacketPersistsAndRetriesAfterRestart(t *testing.T) { + e, key, d := offlineStream(t) + ptx, _ := offlineFunding(t, e, key, d.TemplateID, nil) + events := make(chan clientlib.TransactionEvent) + e.indexer.setTxs(events) + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan struct{}) + go func() { defer close(done); e.svc.watchStreams(ctx) }() + events <- clientlib.TransactionEvent{ArkTx: &clientlib.TxNotification{TxData: clientlib.TxData{Tx: b64(t, ptx)}}} + require.Eventually(t, func() bool { return len(children(t, e, d.ID)) == 1 }, 5*time.Second, 10*time.Millisecond) + cancel() + <-done + child := children(t, e, d.ID)[0] + require.Contains(t, child.Variables, "receiver_program") + require.WithinDuration(t, time.Now().Add(24*time.Hour), *child.ExpiresAt, time.Minute) + e.svc.react(t.Context(), b64(t, ptx)) + require.Len(t, children(t, e, d.ID), 1) + e.indexer.vtxos = append(e.indexer.vtxos, e.vtxoAt(t, ptx.UnsignedTx.TxOut[0].PkScript, 10_000, time.Hour)) + calls := 0 + e.emulator.submitTx = func(string, []string) (string, []string, error) { + calls++ + return "", nil, status.Error(codes.InvalidArgument, "retryable emulator rejection") + } + r := e.restart(t) + r.scan(t) + require.Equal(t, 1, calls) + require.NotNil(t, r.svc.watched[child.ID]) + r = r.restart(t) + r.indexer.serve(e.indexer.vtxos...) + r.scan(t) + require.Equal(t, 2, calls) + require.Len(t, children(t, r, d.ID), 1) +} + +func TestOfflineReceiveMinimumAmount(t *testing.T) { + for _, amount := range []int64{9999, 10_000, 12_000} { + e, key, d := offlineStream(t) + ptx, legacy := offlineFunding(t, e, key, d.TemplateID, nil) + ptx.UnsignedTx.TxOut[0].Value = amount + e.svc.react(t.Context(), b64(t, ptx)) + require.Len(t, children(t, e, d.ID), 1) + child := children(t, e, d.ID)[0] + tmpl, err := e.svc.parseTemplate(t.Context(), d.TemplateID) + require.NoError(t, err) + inst, err := e.svc.newInstance(t.Context(), tmpl, e.svc.cosigners[0], child.Variables, nil, nil) + require.NoError(t, err) + tx := ptx.UnsignedTx + source := &template.Source{Tx: tx, Amount: uint64(amount), Assets: []template.Asset{{ID: bytes.Repeat([]byte{1}, 34), Amount: 7}}} + source.Outpoint.Hash = tx.TxHash() + act, err := inst.Build(t.Context(), []*template.Source{source}, template.Fee{Dust: 330}) + if amount < 10_000 { + require.ErrorIs(t, err, template.ErrIneligible) + require.Nil(t, act) + } else { + require.NoError(t, err) + require.EqualValues(t, amount, act.Tx.UnsignedTx.TxOut[0].Value) + ext, err := extension.NewExtensionFromTx(act.Tx.UnsignedTx) + require.NoError(t, err) + assets := ext.GetAssetPacket() + require.Len(t, assets, 1) + require.Len(t, assets[0].Outputs, 1) + require.EqualValues(t, 0, assets[0].Outputs[0].Vout) + require.EqualValues(t, 7, assets[0].Outputs[0].Amount) + require.Equal(t, "5120"+legacy["receiver_program"], hex.EncodeToString(act.Tx.UnsignedTx.TxOut[0].PkScript)) + } + } +} + +func TestOfflineReceiveRejectsForeignEnvelopes(t *testing.T) { + for _, field := range []string{"template_id", "format", "preimage"} { + t.Run(field, func(t *testing.T) { + e, key, d := offlineStream(t) + ptx, _ := offlineFunding(t, e, key, d.TemplateID, func(v map[string]any) { + if field == "format" || field == "template_id" { + v[field] = "wrong" + } else { + v["values"].(map[string]string)[field] = hex.EncodeToString(make([]byte, 32)) + } + }) + e.svc.react(t.Context(), b64(t, ptx)) + require.Empty(t, children(t, e, d.ID)) + }) + } + for _, name := range []string{"foreign key", "foreign encryption", "malformed JSON", "raw preimage", "missing marker", "foreign marker", "changed payout"} { + t.Run(name, func(t *testing.T) { + e, key, d := offlineStream(t) + ptx, legacy := offlineFunding(t, e, key, d.TemplateID, nil) + if name == "changed payout" { + altered := maps.Clone(legacy) + altered["receiver_program"] = hex.EncodeToString(e.userKey.PubKey().SerializeCompressed()[1:]) + tmpl, err := e.svc.parseTemplate(t.Context(), e.fixture(t, "vhtlc_claim.json")) + require.NoError(t, err) + inst, err := e.svc.newInstance(t.Context(), tmpl, e.svc.cosigners[0], altered, nil, nil) + require.NoError(t, err) + tree, err := txutils.TapTree(inst.Tapscripts(0)).Encode() + require.NoError(t, err) + ptx.UnsignedTx.TxOut[0] = wire.NewTxOut(10_000, inst.PkScript(0)) + ptx.Outputs[0].TaprootTapTree = tree + } else { + body, _ := packets.Find(ptx.UnsignedTx, 4) + switch name { + case "foreign key": + other, _ := hexKey(t) + body = append(body, claimPacket(nil, nil, other.PubKey().SerializeCompressed())...) + case "foreign encryption": + other, _ := hexKey(t) + ct, err := ecies.Encrypt(other.PubKey(), []byte("{}")) + require.NoError(t, err) + body = append(body, claimPacket(ct, nil, nil)...) + case "malformed JSON", "raw preimage": + raw := []byte("{invalid") + if name == "raw preimage" { + raw = make([]byte, 32) + } + ct, err := ecies.Encrypt(key.PubKey(), raw) + require.NoError(t, err) + body = append(body, claimPacket(ct, nil, nil)...) + case "missing marker": + body = body[3+len("delegatee-secrets/v1"):] + case "foreign marker": + body[3] = 'x' + } + setOfflinePacket(t, ptx, body) + } + e.svc.react(t.Context(), b64(t, ptx)) + require.Empty(t, children(t, e, d.ID)) + if name == "changed payout" { + body, _ := packets.Find(ptx.UnsignedTx, 4) + body = append(body, claimPacket(nil, enforcePayTo(t, e.userKey.PubKey().SerializeCompressed()[1:]), nil)...) + setOfflinePacket(t, ptx, body) + e.svc.react(t.Context(), b64(t, ptx)) + require.Len(t, children(t, e, d.ID), 1) + child := children(t, e, d.ID)[0] + tmpl, err := e.svc.parseTemplate(t.Context(), d.TemplateID) + require.NoError(t, err) + inst, err := e.svc.newInstance(t.Context(), tmpl, e.svc.cosigners[0], child.Variables, nil, nil) + require.NoError(t, err) + source := &template.Source{Tx: ptx.UnsignedTx, Amount: 10_000} + source.Outpoint.Hash = ptx.UnsignedTx.TxHash() + act, err := inst.Build(t.Context(), []*template.Source{source}, template.Fee{Dust: 330}) + require.ErrorIs(t, err, template.ErrIneligible) + require.Nil(t, act) + } + }) + } +} + +func TestOfflineReceiveKeepsLegacyStreamSeparate(t *testing.T) { + e, key, d := offlineStream(t) + legacy, err := e.svc.RegisterStreamDelegation(t.Context(), e.trust(t, e.fixture(t, "noninteractiveclaim.json")), nil) + require.NoError(t, err) + e.scan(t) + sealed, _ := offlineFunding(t, e, key, d.TemplateID, nil) + e.svc.react(t.Context(), b64(t, sealed)) + require.Len(t, children(t, e, d.ID), 1) + require.Empty(t, children(t, e, legacy.ID)) + raw, _ := funding(t, e, key.PubKey()) + e.svc.react(t.Context(), b64(t, raw)) + require.Len(t, children(t, e, d.ID), 1) + require.Len(t, children(t, e, legacy.ID), 1) + _, err = e.svc.RegisterDelegation(t.Context(), d.TemplateID, children(t, e, d.ID)[0].Variables, nil) + require.ErrorIs(t, err, ErrInvalidArgs) +} + +func TestOfflineReceiveRejectsAuthenticatedConstraintsBeforeDisclosure(t *testing.T) { + for _, name := range []string{"receiver", "zero minimum", "negative minimum", "excessive minimum", "noncanonical minimum", "missing minimum"} { + t.Run(name, func(t *testing.T) { + e, key, d := offlineStream(t) + ptx, _ := offlineFunding(t, e, key, d.TemplateID, func(envelope map[string]any) { + values := envelope["values"].(map[string]string) + switch name { + case "receiver": + values["receiver_program"] = hex.EncodeToString(e.userKey.PubKey().SerializeCompressed()[1:]) + case "zero minimum": + values["minimum_value"] = "" + case "negative minimum": + values["minimum_value"] = "81" + case "excessive minimum": + values["minimum_value"] = hex.EncodeToString(scriptNum(2100000000000001)) + case "noncanonical minimum": + values["minimum_value"] = "0100" + case "missing minimum": + delete(values, "minimum_value") + } + }) + e.svc.react(t.Context(), b64(t, ptx)) + require.Len(t, children(t, e, d.ID), 1) + child := children(t, e, d.ID)[0] + tmpl, err := e.svc.parseTemplate(t.Context(), d.TemplateID) + require.NoError(t, err) + inst, err := e.svc.newInstance(t.Context(), tmpl, e.svc.cosigners[0], child.Variables, nil, nil) + require.NoError(t, err) + source := &template.Source{Tx: ptx.UnsignedTx, Amount: 10_000} + source.Outpoint.Hash = ptx.UnsignedTx.TxHash() + act, err := inst.Build(t.Context(), []*template.Source{source}, template.Fee{Dust: 330}) + require.ErrorIs(t, err, template.ErrIneligible) + require.Nil(t, act) + e.indexer.vtxos = append(e.indexer.vtxos, e.vtxoAt(t, ptx.UnsignedTx.TxOut[0].PkScript, 10_000, time.Hour)) + calls := 0 + e.emulator.submitTx = func(string, []string) (string, []string, error) { calls++; return "", nil, errBoom } + e.scan(t) + require.Zero(t, calls) + }) + } +} diff --git a/internal/core/application/scale_test.go b/internal/core/application/scale_test.go index fcac627..be840eb 100644 --- a/internal/core/application/scale_test.go +++ b/internal/core/application/scale_test.go @@ -1,6 +1,7 @@ package application import ( + "fmt" "testing" "time" @@ -10,6 +11,24 @@ import ( "github.com/stretchr/testify/require" ) +func BenchmarkVtxoCache50000Scripts(b *testing.B) { + i := &cacheIndexer{coins: map[string][]clientlib.Vtxo{}} + s := &service{indexer: i, wake: make(chan struct{}, 1), incrementalRefresh: true} + scripts := make([]string, 50000) + for n := range scripts { + scripts[n] = fmt.Sprintf("5120%064x", n) + } + _, err := s.scanVtxos(b.Context(), scripts) + require.NoError(b, err) + b.ResetTimer() + for range b.N { + s.invalidateScripts(scripts[0]) + _, err := s.scanVtxos(b.Context(), scripts) + require.NoError(b, err) + i.requests = i.requests[:0] + } +} + func BenchmarkScan5000(b *testing.B) { env := newTestEnv(b) manyDelegations(b, env, 5000) diff --git a/internal/core/application/service.go b/internal/core/application/service.go index 2347568..5b5ca1e 100644 --- a/internal/core/application/service.go +++ b/internal/core/application/service.go @@ -238,15 +238,22 @@ type service struct { failing map[string]*failure // coins whose last attempt failed // the scan alone - held map[string]bool // coins of the daemon's settlements that may still land, as of the scan - entries []planEntry // the batch lane's future intents, as the scan found them - found map[int64][]scannedCoin - backoff time.Duration // grows while an upstream throttles the scan - plan *plan // published under mu - refusedSince map[string]time.Time // pending txs arkd refuses, since the first refusal - watched map[int64]*watched // nil for a delegation of another key - sub *subscription - sampled map[int64]bool // inFlight as the scan began + held map[string]bool // coins of the daemon's settlements that may still land, as of the scan + entries []planEntry // the batch lane's future intents, as the scan found them + found map[int64][]scannedCoin + backoff time.Duration + plan *plan // published under mu + refusedSince map[string]time.Time // pending txs arkd refuses, since the first refusal + watched map[int64]*watched // nil for a delegation of another key + subs []*subscription + sampled map[int64]bool // inFlight as the scan began + vtxoCache map[string][]clientlib.Vtxo + lastFullRefresh time.Time + incrementalRefresh bool + + vtxoDirtyMu sync.Mutex + vtxoDirty map[string]struct{} + vtxoDirtyAll bool // mu guards what the last completed scan saw mu sync.Mutex @@ -459,16 +466,19 @@ func (s *service) Start() { defer close(s.stopped) timer := time.NewTimer(s.pollInterval) defer timer.Stop() + full := true for { - err := s.scan(ctx) + err := s.scanWithRefresh(ctx, full) scanned := time.Now() delay, gap := s.scanDelays(err, scanned) timer.Reset(delay) select { case <-ctx.Done(): case <-timer.C: + full = true continue case <-s.wake: + full = false select { case <-ctx.Done(): case <-time.After(gap - time.Since(scanned)): @@ -493,6 +503,9 @@ func (s *service) scanDelays(err error, now time.Time) (delay, gap time.Duration s.backoff = 0 } delay, gap = max(s.nextScanDelay(now), s.backoff), max(minScanGap, s.backoff) + if s.backoff == 0 && !s.lastFullRefresh.IsZero() { + delay = min(delay, max(minScanGap, s.lastFullRefresh.Add(s.pollInterval).Sub(now))) + } s.mu.Lock() s.nextScan, s.throttled = now.Add(delay), s.backoff > 0 s.mu.Unlock() @@ -591,7 +604,7 @@ func (s *service) registerWatch( ExpiresAt: expiresAt, DelegatePubKey: s.cosigners[0].pubKey, Slots: slots, }, limit) if err == nil && tmpl.Type() != template.Intent { - s.wakeScan() // the coin may be there already + s.invalidateScripts(slots[0].Script) } return d, err } @@ -699,7 +712,7 @@ func (s *service) ResumeDelegation(ctx context.Context, id int64) error { err := s.repo.Resume(ctx, id) s.registerMu.Unlock() if err == nil { - s.wakeScan() + s.invalidateAllScripts() } return err } @@ -1029,6 +1042,19 @@ func (s *service) spendableVtxos(ctx context.Context, scripts []string) (map[str } func (s *service) scan(ctx context.Context) error { + return s.scanWithRefresh(ctx, true) +} + +func (s *service) scanWithRefresh(ctx context.Context, full bool) (err error) { + lastFullRefresh := s.lastFullRefresh + s.incrementalRefresh = !full + defer func() { + s.incrementalRefresh = false + if err != nil { + s.lastFullRefresh = lastFullRefresh + s.markScriptsStale() + } + }() s.collectUntil = time.Time{} s.sampleLanes() now := time.Now() diff --git a/internal/core/application/spend.go b/internal/core/application/spend.go index 899c311..ae00152 100644 --- a/internal/core/application/spend.go +++ b/internal/core/application/spend.go @@ -63,6 +63,9 @@ func (s *service) RegisterSpend( ExpiresAt: &expiresAt, DelegatePubKey: s.cosigners[0].pubKey, Slots: slots, }, s.limits.MaxDelegations) if err == nil { + for _, slot := range d.Slots { + s.markScriptsDirty(slot.Script) + } s.wakeScan() } return d, err diff --git a/internal/core/application/spend_test.go b/internal/core/application/spend_test.go index 104f92a..178afa6 100644 --- a/internal/core/application/spend_test.go +++ b/internal/core/application/spend_test.go @@ -2,6 +2,7 @@ package application import ( "encoding/hex" + "slices" "testing" "time" @@ -167,3 +168,20 @@ func (e *testEnv) spendRequest(t *testing.T) spendRequest { r.sign(t, e.userKey) return r } + +func TestRegisterSpendRefreshesPreviouslyEmptyScript(t *testing.T) { + e := newTestEnv(t) + req := e.spendRequest(t) + coins := slices.Clone(e.indexer.vtxos) + script := coins[0].Script + e.svc.incrementalRefresh = true + e.indexer.serve() + _, err := e.svc.scanVtxos(t.Context(), []string{script}) + require.NoError(t, err) + e.indexer.serve(coins...) + _, err = req.register(t, e) + require.NoError(t, err) + got, err := e.svc.scanVtxos(t.Context(), []string{script}) + require.NoError(t, err) + require.Len(t, got[script], 2) +} diff --git a/internal/core/application/stream.go b/internal/core/application/stream.go index 73bc331..c44e0a7 100644 --- a/internal/core/application/stream.go +++ b/internal/core/application/stream.go @@ -3,6 +3,7 @@ package application import ( "context" "errors" + "fmt" "maps" "slices" "strings" @@ -69,7 +70,7 @@ func (s *service) streamSet() []stream { const streamRetry = time.Second -// watchStreams follows arkd's transactions while a stream delegation is active. +// watchStreams follows matching packets while a stream delegation is active. func (s *service) watchStreams(ctx context.Context) { failing := false for { @@ -103,26 +104,74 @@ func (s *service) watchStreams(ctx context.Context) { func (s *service) readTransactions(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) defer cancel() - events, stop, err := s.ark.GetTransactionsStream(ctx) + select { + case <-s.streamsChanged: + default: + } + filters, err := s.streamFilters() + if err != nil { + return err + } + if len(filters) == 0 { + return nil + } + id, events, stop, err := s.indexer.NewTransactionSubscription(ctx, filters) if err != nil { return err } defer stop() + type filterUpdate struct { + filters []string + err error + } + var updating <-chan filterUpdate + pending, retryUpdate := false, false + retry := time.NewTicker(streamRetry) + defer retry.Stop() for { + if pending && updating == nil { + pending = false + next, err := s.streamFilters() + switch { + case err != nil: + log.WithError(err).Warn("prepare transaction filters") + retryUpdate = true + case len(next) == 0: + return nil + case slices.Equal(filters, next) && !retryUpdate: + retryUpdate = false + default: + result := make(chan filterUpdate, 1) + updating = result + go func() { + updateCtx, cancelUpdate := context.WithTimeout(ctx, 10*time.Second) + defer cancelUpdate() + err := s.indexer.UpdateTransactionSubscription(updateCtx, id, next) + result <- filterUpdate{filters: next, err: err} + }() + } + } select { case <-ctx.Done(): return nil case <-s.streamsChanged: - if len(s.streamSet()) == 0 { - return nil + pending = true + case <-retry.C: + pending = pending || retryUpdate + case update := <-updating: + updating = nil + retryUpdate = update.err != nil + if update.err != nil { + log.WithError(update.err).Warn("update transaction filters") + } else { + filters = update.filters } case ev, ok := <-events: if !ok { return nil } if ev.Err != nil { - log.WithError(ev.Err).Debug("transaction stream") - continue + return ev.Err } if ev.ArkTx != nil { s.react(ctx, ev.ArkTx.Tx) @@ -131,6 +180,32 @@ func (s *service) readTransactions(ctx context.Context) error { } } +func (s *service) streamFilters() ([]string, error) { + unique := map[string]struct{}{} + for _, st := range s.streamSet() { + unique[st.trigger.Filter(s.triggerKeys())] = struct{}{} + } + var filters []string + expression := "" + for _, filter := range slices.Sorted(maps.Keys(unique)) { + if len(expression)+len(filter)+6 > 64<<10 { + filters = append(filters, expression) + expression = "" + } + if expression != "" { + expression += " || " + } + expression += "(" + filter + ")" + } + if expression != "" { + filters = append(filters, expression) + } + if len(filters) > 64 { + return nil, fmt.Errorf("stream templates exceed the indexer's 64 filter limit") + } + return filters, nil +} + // react spawns a watch for every output of a streamed transaction that a trigger binds and its template locks. func (s *service) react(ctx context.Context, raw string) { ptx, err := psbt.NewFromRawBytes(strings.NewReader(raw), true) diff --git a/internal/core/application/stream_test.go b/internal/core/application/stream_test.go index b0832bc..3afe5f9 100644 --- a/internal/core/application/stream_test.go +++ b/internal/core/application/stream_test.go @@ -5,6 +5,8 @@ import ( "encoding/binary" "encoding/hex" "fmt" + "slices" + "sync" "testing" "time" @@ -153,7 +155,7 @@ func TestWatchStreamsReopens(t *testing.T) { e, secrets, d := streamEnv(t) e.scan(t) first := make(chan clientlib.TransactionEvent) - e.ark.setTxs(first) + e.indexer.setTxs(first) ctx, cancel := context.WithCancel(t.Context()) done := make(chan struct{}) @@ -165,9 +167,9 @@ func TestWatchStreamsReopens(t *testing.T) { first <- clientlib.TransactionEvent{ArkTx: &clientlib.TxNotification{TxData: clientlib.TxData{Tx: b64(t, ptx)}}} require.Eventually(t, func() bool { return len(children(t, e, d.ID)) == 1 }, 5*time.Second, 10*time.Millisecond) - e.ark.setTxs(make(chan clientlib.TransactionEvent)) + e.indexer.setTxs(make(chan clientlib.TransactionEvent)) close(first) - require.Eventually(t, e.ark.opened(2), 5*time.Second, 10*time.Millisecond, "reopened after the stream ended") + require.Eventually(t, e.indexer.transactionsOpened(2), 5*time.Second, 10*time.Millisecond, "reopened after the stream ended") cancel() select { @@ -179,9 +181,9 @@ func TestWatchStreamsReopens(t *testing.T) { func TestStopEndsTheWatcher(t *testing.T) { e, _, _ := streamEnv(t) - e.ark.setTxs(make(chan clientlib.TransactionEvent)) + e.indexer.setTxs(make(chan clientlib.TransactionEvent)) e.svc.Start() - require.Eventually(t, e.ark.opened(1), 5*time.Second, 10*time.Millisecond, "the first scan opens the stream") + require.Eventually(t, e.indexer.transactionsOpened(1), 5*time.Second, 10*time.Millisecond, "the first scan opens the stream") stopped := make(chan struct{}) go func() { e.svc.Stop() @@ -343,3 +345,112 @@ func children(t *testing.T, e *testEnv, parent int64) []domain.Delegation { } return out } + +func TestStreamSubscriptionUsesRecipientFilter(t *testing.T) { + e, key, d := offlineStream(t) + events := make(chan clientlib.TransactionEvent) + e.indexer.setTxs(events) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan struct{}) + go func() { defer close(done); e.svc.watchStreams(ctx) }() + require.Eventually(t, e.indexer.transactionsOpened(1), 5*time.Second, time.Millisecond) + e.indexer.mu.Lock() + filters := slices.Clone(e.indexer.filters[0]) + e.indexer.mu.Unlock() + require.Len(t, filters, 1) + require.Contains(t, filters[0], "hasPacket(tx.extension, 4)") + require.Contains(t, filters[0], hex.EncodeToString(key.PubKey().SerializeCompressed())) + require.Contains(t, filters[0], hex.EncodeToString([]byte("delegatee-secrets/v1"))) + require.NoError(t, e.svc.CancelDelegationByID(t.Context(), d.ID)) + e.scan(t) + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("filtered stream did not stop") + } +} + +func TestStreamFilterUpdateKeepsReceiving(t *testing.T) { + e, key, d := offlineStream(t) + events := make(chan clientlib.TransactionEvent) + e.indexer.setTxs(events) + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + e.indexer.updateTransactions = func() error { once.Do(func() { close(entered); <-release }); return nil } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan struct{}) + go func() { defer close(done); e.svc.watchStreams(ctx) }() + require.Eventually(t, e.indexer.transactionsOpened(1), time.Second, time.Millisecond) + _, err := e.svc.RegisterStreamDelegation(t.Context(), e.trust(t, e.fixture(t, "noninteractiveclaim.json")), nil) + require.NoError(t, err) + e.scan(t) + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("filter update did not start") + } + ptx, _ := offlineFunding(t, e, key, d.TemplateID, nil) + events <- clientlib.TransactionEvent{ArkTx: &clientlib.TxNotification{TxData: clientlib.TxData{Tx: b64(t, ptx)}}} + require.Eventually(t, func() bool { return len(children(t, e, d.ID)) == 1 }, time.Second, time.Millisecond) + close(release) + require.Eventually(t, func() bool { e.indexer.mu.Lock(); defer e.indexer.mu.Unlock(); return len(e.indexer.filters) == 2 }, time.Second, time.Millisecond) + require.True(t, e.indexer.transactionsOpened(1)(), "filter updates keep the original subscription") + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("watcher did not stop") + } +} + +func TestStreamFilterReconcilesAmbiguousUpdate(t *testing.T) { + e, key, d := offlineStream(t) + events := make(chan clientlib.TransactionEvent) + e.indexer.setTxs(events) + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + attempt := 0 + e.indexer.updateTransactions = func() error { + attempt++ + once.Do(func() { close(entered); <-release }) + if attempt == 1 { + return errBoom + } + return nil + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan struct{}) + go func() { defer close(done); e.svc.watchStreams(ctx) }() + require.Eventually(t, e.indexer.transactionsOpened(1), time.Second, time.Millisecond) + other, err := e.svc.RegisterStreamDelegation(t.Context(), e.trust(t, e.fixture(t, "noninteractiveclaim.json")), nil) + require.NoError(t, err) + e.scan(t) + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("filter update did not start") + } + ptx, _ := offlineFunding(t, e, key, d.TemplateID, nil) + events <- clientlib.TransactionEvent{ArkTx: &clientlib.TxNotification{TxData: clientlib.TxData{Tx: b64(t, ptx)}}} + require.Eventually(t, func() bool { return len(children(t, e, d.ID)) == 1 }, time.Second, time.Millisecond) + require.NoError(t, e.svc.CancelDelegationByID(t.Context(), other.ID)) + e.scan(t) + require.Eventually(t, func() bool { return len(e.svc.streamsChanged) == 0 }, time.Second, time.Millisecond) + close(release) + require.Eventually(t, func() bool { e.indexer.mu.Lock(); defer e.indexer.mu.Unlock(); return len(e.indexer.filters) == 2 }, time.Second, time.Millisecond) + e.indexer.mu.Lock() + updates := e.indexer.txUpdates + e.indexer.mu.Unlock() + require.Equal(t, 2, updates) + require.True(t, e.indexer.transactionsOpened(1)(), "filter updates keep the original subscription") + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("watcher did not stop") + } +} diff --git a/internal/core/application/subscription.go b/internal/core/application/subscription.go index 2b08184..a5796d0 100644 --- a/internal/core/application/subscription.go +++ b/internal/core/application/subscription.go @@ -15,6 +15,8 @@ import ( // woken scans are at least this far apart const minScanGap = time.Second +const maxSubscriptionScripts = 1000 + // subscription wakes the scan when a coin reaches a watched script. type subscription struct { id string @@ -47,73 +49,140 @@ func (s *service) wakeScripts() []string { return scripts } -// syncSubscription never fails a scan: without it a claim waits for the poll. +// syncSubscription keeps each shard within the indexer's 1000-script limit. func (s *service) syncSubscription(ctx context.Context, want []string) { wanted := make(map[string]struct{}, len(want)) for _, script := range want { wanted[script] = struct{}{} } - dead := s.sub != nil && s.sub.dead.Load() - if s.sub != nil && (dead || len(wanted) == 0) { - s.sub.stop() - s.sub = nil + var live []*subscription + var plans []map[string]struct{} + assigned := make(map[string]struct{}, len(wanted)) + dead := false + for _, sub := range s.subs { + if sub.dead.Load() { + sub.stop() + dead = true + continue + } + kept := map[string]struct{}{} + for script := range sub.scripts { + if _, ok := wanted[script]; ok { + kept[script] = struct{}{} + assigned[script] = struct{}{} + } + } + live = append(live, sub) + plans = append(plans, kept) } - if len(wanted) == 0 { - return + remaining := make([]string, 0, len(wanted)-len(assigned)) + for script := range wanted { + if _, ok := assigned[script]; !ok { + remaining = append(remaining, script) + } } - if s.sub != nil { - var add, remove []string - for script := range wanted { - if _, ok := s.sub.scripts[script]; !ok { - add = append(add, script) + slices.Sort(remaining) + for n := 0; len(remaining) > 0; n++ { + if n == len(plans) { + plans = append(plans, map[string]struct{}{}) + } + count := min(maxSubscriptionScripts-len(plans[n]), len(remaining)) + for _, script := range remaining[:count] { + plans[n][script] = struct{}{} + } + remaining = remaining[count:] + } + var next []*subscription + for n, plan := range plans { + var sub *subscription + if n < len(live) { + sub = live[n] + } + if len(plan) == 0 { + if sub != nil { + sub.stop() } + continue } - for script := range s.sub.scripts { - if _, ok := wanted[script]; !ok { - remove = append(remove, script) + if sub != nil { + var add, remove []string + for script := range plan { + if _, ok := sub.scripts[script]; !ok { + add = append(add, script) + } + } + for script := range sub.scripts { + if _, ok := plan[script]; !ok { + remove = append(remove, script) + } + } + if len(add)+len(remove) == 0 { + next = append(next, sub) + continue + } + if err := s.indexer.UpdateSubscription(ctx, sub.id, add, remove); err == nil { + sub.scripts = plan + next = append(next, sub) + if len(add) > 0 { + s.invalidateScripts(add...) + } + continue + } else { + log.WithError(err).Warn("update the script subscription") + s.invalidateAllScripts() + sub.stop() } } - if len(add)+len(remove) == 0 { - return + id, events, stop, err := s.indexer.NewSubscription(ctx, slices.Collect(maps.Keys(plan))) + if err != nil { + log.WithError(err).Warn("subscribe to watched scripts") + continue } - err := s.indexer.UpdateSubscription(ctx, s.sub.id, add, remove) - if err == nil { - s.sub.scripts = wanted - if len(add) > 0 { - s.wakeScan() // a coin may have reached an added script unseen - } - return + sub = &subscription{id: id, scripts: plan, stop: stop} + next = append(next, sub) + go s.follow(sub, events) + if !dead { + s.invalidateScripts(slices.Collect(maps.Keys(plan))...) } - log.WithError(err).Warn("update the script subscription") - s.sub.stop() - s.sub = nil - } - id, events, stop, err := s.indexer.NewSubscription(ctx, slices.Collect(maps.Keys(wanted))) - if err != nil { - log.WithError(err).Warn("subscribe to watched scripts") - return - } - s.sub = &subscription{id: id, scripts: wanted, stop: stop} - go s.follow(s.sub, events) - // a stream that keeps ending would otherwise scan in a loop - if !dead { - s.wakeScan() // a coin may have arrived before the stream opened } + s.subs = next } func (s *service) follow(sub *subscription, events <-chan clientlib.ScriptEvent) { - defer sub.dead.Store(true) + defer func() { + sub.dead.Store(true) + s.markScriptsStale() + }() for ev := range events { if ev.Err != nil { log.WithError(ev.Err).Warn("script subscription") } - if wakes(ev) { - s.wakeScan() + if ev.Connection != nil || ev.Err != nil { + s.invalidateAllScripts() + continue + } + if !wakes(ev) { + continue + } + scripts := slices.Clone(ev.Data.Scripts) + for _, v := range ev.Data.NewVtxos { + if v.Script != "" { + scripts = append(scripts, v.Script) + } + } + for _, v := range ev.Data.SpentVtxos { + if v.Script != "" { + scripts = append(scripts, v.Script) + } + } + if len(scripts) == 0 { + s.invalidateAllScripts() + } else { + s.invalidateScripts(scripts...) } } } -// wakes: a new coin, or a connection change that may have hidden one. func wakes(ev clientlib.ScriptEvent) bool { - return ev.Connection != nil || (ev.Data != nil && len(ev.Data.NewVtxos) > 0) + return ev.Connection != nil || (ev.Data != nil && len(ev.Data.NewVtxos)+len(ev.Data.SpentVtxos) > 0) } diff --git a/internal/core/application/subscription_test.go b/internal/core/application/subscription_test.go index 2f08e73..cb9a829 100644 --- a/internal/core/application/subscription_test.go +++ b/internal/core/application/subscription_test.go @@ -1,6 +1,8 @@ package application import ( + "fmt" + "slices" "sync/atomic" "testing" "testing/synctest" @@ -109,13 +111,13 @@ func TestSubscriptionFollowsDirectWatches(t *testing.T) { e.indexer.serve() e.scan(t) require.Empty(t, e.indexer.subscribedScripts()) - require.Nil(t, e.svc.sub, "no swap watch, no subscription") + require.Empty(t, e.svc.subs, "no swap watch, no subscription") } func TestScriptEventWakesTheScan(t *testing.T) { require.True(t, wakes(clientlib.ScriptEvent{Data: &clientlib.ScriptEventData{NewVtxos: []clientlib.Vtxo{{}}}})) require.True(t, wakes(clientlib.ScriptEvent{Connection: &clientlib.StreamConnectionEvent{}}), "events may have been missed") - require.False(t, wakes(clientlib.ScriptEvent{Data: &clientlib.ScriptEventData{SpentVtxos: []clientlib.Vtxo{{}}}}), "the daemon's own claim") + require.True(t, wakes(clientlib.ScriptEvent{Data: &clientlib.ScriptEventData{SpentVtxos: []clientlib.Vtxo{{}}}})) require.False(t, wakes(clientlib.ScriptEvent{Err: errBoom})) e, _, _ := claimEnv(t) @@ -127,14 +129,14 @@ func TestScriptEventWakesTheScan(t *testing.T) { func TestSubscriptionReopensAfterItsStreamEnds(t *testing.T) { e, _, _ := claimEnv(t) e.openSubscription(t) - sub := e.svc.sub + sub := e.svc.subs[0] e.indexer.closeSubscription() require.Eventually(t, sub.dead.Load, time.Second, time.Millisecond) e.indexer.serve() e.scan(t) require.Equal(t, 2, e.indexer.subscriptions) - require.NotSame(t, sub, e.svc.sub) + require.NotSame(t, sub, e.svc.subs[0]) require.Empty(t, e.svc.wake, "a stream that keeps ending must not scan in a loop") } @@ -144,12 +146,12 @@ func TestSubscribeFailureKeepsScanning(t *testing.T) { e.indexer.serve() e.scan(t) require.False(t, e.svc.Status().LastScan.IsZero()) - require.Nil(t, e.svc.sub) + require.Empty(t, e.svc.subs) e.indexer.subErr = nil e.indexer.serve() e.scan(t) - require.NotNil(t, e.svc.sub) + require.NotEmpty(t, e.svc.subs) } func TestOpeningASubscriptionWakesTheScan(t *testing.T) { @@ -157,7 +159,7 @@ func TestOpeningASubscriptionWakesTheScan(t *testing.T) { <-e.svc.wake // the registration's e.indexer.serve() e.scan(t) - require.NotNil(t, e.svc.sub) + require.NotEmpty(t, e.svc.subs) require.Len(t, e.svc.wake, 1, "a coin may have arrived before the stream opened") } @@ -179,15 +181,15 @@ func TestGrowingASubscriptionWakesTheScan(t *testing.T) { func TestFailedUpdateReopensInTheSameScan(t *testing.T) { e := newTestEnv(t) e.svc.syncSubscription(t.Context(), []string{"a"}) - old := e.svc.sub + old := e.svc.subs[0] e.indexer.updateErr = errBoom e.svc.syncSubscription(t.Context(), []string{"a", "b"}) require.Equal(t, 2, e.indexer.subscriptions) - require.NotNil(t, e.svc.sub) - require.NotSame(t, old, e.svc.sub) + require.NotEmpty(t, e.svc.subs) + require.NotSame(t, old, e.svc.subs[0]) require.Equal(t, []string{"a", "b"}, e.indexer.subscribedScripts()) - require.Len(t, e.svc.sub.scripts, 2) + require.Len(t, e.svc.subs[0].scripts, 2) } func TestSubscriptionOpensEachScriptOnce(t *testing.T) { @@ -196,3 +198,48 @@ func TestSubscriptionOpensEachScriptOnce(t *testing.T) { require.Len(t, e.indexer.opened, 1) require.ElementsMatch(t, []string{"a", "b"}, e.indexer.opened[0]) } + +func TestSubscriptionShardsAtServerLimit(t *testing.T) { + e := newTestEnv(t) + scripts := make([]string, 50000) + for n := range scripts { + scripts[n] = fmt.Sprintf("script-%05d", n) + } + e.svc.syncSubscription(t.Context(), scripts) + require.Len(t, e.svc.subs, 50) + require.Len(t, e.indexer.subscribedScripts(), len(scripts)) + ids := make([]string, len(e.svc.subs)) + for n, sub := range e.svc.subs { + require.Len(t, sub.scripts, maxSubscriptionScripts) + ids[n] = sub.id + } + e.svc.syncSubscription(t.Context(), scripts) + require.Equal(t, 50, e.indexer.subscriptions) + e.svc.syncSubscription(t.Context(), append(slices.Clone(scripts), "new-script")) + require.Len(t, e.svc.subs, 51) + for n, id := range ids { + require.Equal(t, id, e.svc.subs[n].id) + } + e.svc.syncSubscription(t.Context(), scripts[:999]) + require.Len(t, e.svc.subs, 1) + require.ElementsMatch(t, scripts[:999], e.indexer.subscribedScripts()) + e.svc.syncSubscription(t.Context(), nil) + require.Empty(t, e.svc.subs) + require.Empty(t, e.indexer.subscribedScripts()) +} + +func TestDeadSubscriptionReopensOnlyItsShard(t *testing.T) { + e := newTestEnv(t) + scripts := make([]string, 2000) + for n := range scripts { + scripts[n] = fmt.Sprintf("script-%05d", n) + } + e.svc.syncSubscription(t.Context(), scripts) + first, second := e.svc.subs[0], e.svc.subs[1] + first.stop() + require.Eventually(t, first.dead.Load, time.Second, time.Millisecond) + e.svc.syncSubscription(t.Context(), scripts) + require.Equal(t, 3, e.indexer.subscriptions) + require.Same(t, second, e.svc.subs[0]) + require.ElementsMatch(t, scripts, e.indexer.subscribedScripts()) +} diff --git a/internal/core/application/vtxo_cache.go b/internal/core/application/vtxo_cache.go new file mode 100644 index 0000000..64b3e04 --- /dev/null +++ b/internal/core/application/vtxo_cache.go @@ -0,0 +1,86 @@ +package application + +import ( + "context" + "maps" + "time" + + clientlib "github.com/arkade-os/arkd/pkg/client-lib" +) + +func (s *service) invalidateScripts(scripts ...string) { + s.markScriptsDirty(scripts...) + s.wakeScan() +} + +func (s *service) markScriptsDirty(scripts ...string) { + s.vtxoDirtyMu.Lock() + if s.vtxoDirty == nil { + s.vtxoDirty = make(map[string]struct{}) + } + for _, script := range scripts { + if script != "" { + s.vtxoDirty[script] = struct{}{} + } + } + s.vtxoDirtyMu.Unlock() +} + +func (s *service) invalidateAllScripts() { + s.markScriptsStale() + s.wakeScan() +} + +func (s *service) markScriptsStale() { + s.vtxoDirtyMu.Lock() + s.vtxoDirtyAll = true + s.vtxoDirtyMu.Unlock() +} + +func (s *service) scanVtxos(ctx context.Context, scripts []string) (map[string][]clientlib.Vtxo, error) { + s.vtxoDirtyMu.Lock() + dirty, full := s.vtxoDirty, s.vtxoDirtyAll + s.vtxoDirty, s.vtxoDirtyAll = nil, false + s.vtxoDirtyMu.Unlock() + full = full || !s.incrementalRefresh || s.lastFullRefresh.IsZero() || + (s.pollInterval > 0 && time.Since(s.lastFullRefresh) >= s.pollInterval) + wanted := make(map[string]struct{}, len(scripts)) + var refresh []string + for _, script := range scripts { + if _, duplicate := wanted[script]; duplicate { + continue + } + wanted[script] = struct{}{} + _, cached := s.vtxoCache[script] + _, changed := dirty[script] + if full || !cached || changed { + refresh = append(refresh, script) + } + } + coins, err := s.spendableVtxos(ctx, refresh) + if err != nil { + s.vtxoDirtyMu.Lock() + if s.vtxoDirty == nil { + s.vtxoDirty = make(map[string]struct{}) + } + maps.Copy(s.vtxoDirty, dirty) + s.vtxoDirtyAll = s.vtxoDirtyAll || full + s.vtxoDirtyMu.Unlock() + return nil, err + } + if s.vtxoCache == nil { + s.vtxoCache = make(map[string][]clientlib.Vtxo, len(scripts)) + } + for _, script := range refresh { + s.vtxoCache[script] = coins[script] + } + for script := range s.vtxoCache { + if _, ok := wanted[script]; !ok { + delete(s.vtxoCache, script) + } + } + if full { + s.lastFullRefresh = time.Now() + } + return s.vtxoCache, nil +} diff --git a/internal/core/application/vtxo_cache_test.go b/internal/core/application/vtxo_cache_test.go new file mode 100644 index 0000000..fe84065 --- /dev/null +++ b/internal/core/application/vtxo_cache_test.go @@ -0,0 +1,258 @@ +package application + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "testing/synctest" + "time" + + clientlib "github.com/arkade-os/arkd/pkg/client-lib" + "github.com/arkade-os/delegatee/internal/core/ports" + "github.com/stretchr/testify/require" +) + +type cacheIndexer struct { + ports.Indexer + mu sync.Mutex + coins map[string][]clientlib.Vtxo + requests [][]string + err error + onRequest func() +} + +func (i *cacheIndexer) requestCount() int { + i.mu.Lock() + defer i.mu.Unlock() + return len(i.requests) +} + +func (i *cacheIndexer) GetVtxos(_ context.Context, opts ...clientlib.GetVtxosOption) (*clientlib.VtxosResponse, error) { + o, err := clientlib.ApplyGetVtxosOptions(opts...) + if err != nil { + return nil, err + } + i.mu.Lock() + i.requests = append(i.requests, append([]string(nil), o.Scripts...)) + result := &clientlib.VtxosResponse{} + for _, script := range o.Scripts { + result.Vtxos = append(result.Vtxos, i.coins[script]...) + } + err, hook := i.err, i.onRequest + i.mu.Unlock() + if hook != nil { + hook() + } + return result, err +} + +func TestVtxoCacheRefresh(t *testing.T) { + i := &cacheIndexer{coins: map[string][]clientlib.Vtxo{"a": {{Script: "a", Amount: 1}}}} + s := &service{indexer: i, wake: make(chan struct{}, 1), incrementalRefresh: true} + got, err := s.scanVtxos(t.Context(), []string{"a", "b"}) + require.NoError(t, err) + require.Equal(t, uint64(1), got["a"][0].Amount) + require.Contains(t, got, "b") + i.coins["a"] = []clientlib.Vtxo{{Script: "a", Amount: 2}} + got, err = s.scanVtxos(t.Context(), []string{"a", "b"}) + require.NoError(t, err) + require.Len(t, i.requests, 1) + require.Equal(t, uint64(1), got["a"][0].Amount) + for range 100 { + s.invalidateScripts("a") + } + require.Len(t, s.wake, 1) + got, err = s.scanVtxos(t.Context(), []string{"a", "b"}) + require.NoError(t, err) + require.Equal(t, uint64(2), got["a"][0].Amount) + require.Equal(t, []string{"a"}, i.requests[1]) + i.coins["a"] = nil + s.invalidateAllScripts() + got, err = s.scanVtxos(t.Context(), []string{"a", "c"}) + require.NoError(t, err) + require.Empty(t, got["a"]) + require.NotContains(t, got, "b") + require.Contains(t, got, "c") +} + +func TestVtxoCacheInvalidationDuringRequest(t *testing.T) { + for _, fail := range []bool{false, true} { + t.Run(fmt.Sprint(fail), func(t *testing.T) { + i := &cacheIndexer{coins: map[string][]clientlib.Vtxo{}} + s := &service{indexer: i, wake: make(chan struct{}, 1), incrementalRefresh: true} + _, err := s.scanVtxos(t.Context(), []string{"a", "b"}) + require.NoError(t, err) + s.invalidateScripts("a") + i.onRequest = func() { s.invalidateScripts("b") } + if fail { + i.err = errors.New("offline") + } + _, err = s.scanVtxos(t.Context(), []string{"a", "b"}) + if fail { + require.Error(t, err) + } else { + require.NoError(t, err) + } + i.onRequest, i.err = nil, nil + _, err = s.scanVtxos(t.Context(), []string{"a", "b"}) + require.NoError(t, err) + if fail { + require.ElementsMatch(t, []string{"a", "b"}, i.requests[2]) + } else { + require.Equal(t, []string{"b"}, i.requests[2]) + } + }) + } +} + +func TestVtxoCachePeriodicRefresh(t *testing.T) { + i := &cacheIndexer{coins: map[string][]clientlib.Vtxo{}} + s := &service{indexer: i, wake: make(chan struct{}, 1), incrementalRefresh: true, pollInterval: time.Minute} + _, err := s.scanVtxos(t.Context(), []string{"a"}) + require.NoError(t, err) + i.coins["a"] = []clientlib.Vtxo{{Script: "a", Amount: 3}} + s.lastFullRefresh = time.Now().Add(-2 * time.Minute) + got, err := s.scanVtxos(t.Context(), []string{"a"}) + require.NoError(t, err) + require.Equal(t, uint64(3), got["a"][0].Amount) + require.WithinDuration(t, time.Now(), s.lastFullRefresh, time.Second) +} + +func TestVtxoCacheFullInvalidationAndFailedRefresh(t *testing.T) { + i := &cacheIndexer{coins: map[string][]clientlib.Vtxo{"a": {{Script: "a", Amount: 1}}}} + s := &service{indexer: i, wake: make(chan struct{}, 1), incrementalRefresh: true} + scripts := []string{"a", "b"} + _, err := s.scanVtxos(t.Context(), scripts) + require.NoError(t, err) + last := s.lastFullRefresh + i.coins["a"] = []clientlib.Vtxo{{Script: "a", Amount: 2}} + i.err = errors.New("offline") + s.markScriptsStale() + got, err := s.scanVtxos(t.Context(), scripts) + require.Error(t, err) + require.Nil(t, got) + require.Equal(t, last, s.lastFullRefresh) + require.Equal(t, uint64(1), s.vtxoCache["a"][0].Amount) + i.err = nil + i.onRequest = s.markScriptsStale + got, err = s.scanVtxos(t.Context(), scripts) + require.NoError(t, err) + require.Equal(t, uint64(2), got["a"][0].Amount) + i.onRequest = nil + _, err = s.scanVtxos(t.Context(), scripts) + require.NoError(t, err) + require.Len(t, i.requests, 4) + for _, request := range i.requests { + require.ElementsMatch(t, scripts, request) + } +} + +func TestVtxoCache50000Scripts(t *testing.T) { + i := &cacheIndexer{coins: map[string][]clientlib.Vtxo{}} + s := &service{indexer: i, wake: make(chan struct{}, 1), incrementalRefresh: true} + scripts := make([]string, 50000) + for n := range scripts { + scripts[n] = fmt.Sprintf("5120%064x", n) + } + _, err := s.scanVtxos(t.Context(), scripts) + require.NoError(t, err) + require.Len(t, i.requests, 500) + s.invalidateScripts(scripts[0], scripts[49999]) + _, err = s.scanVtxos(t.Context(), scripts) + require.NoError(t, err) + require.Len(t, i.requests, 501) + require.ElementsMatch(t, []string{scripts[0], scripts[49999]}, i.requests[500]) +} + +func TestWakeScansCannotPostponeFullReconciliation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + e := newTestEnv(t) + manyDelegations(t, e, 2) + i := &cacheIndexer{Indexer: e.indexer, coins: map[string][]clientlib.Vtxo{}} + for _, v := range e.indexer.vtxos { + i.coins[v.Script] = []clientlib.Vtxo{v} + } + e.svc.indexer = i + e.svc.pollInterval = 5 * time.Second + e.svc.Start() + synctest.Wait() + require.Equal(t, 1, i.requestCount()) + first := e.svc.lastFullRefresh + i.mu.Lock() + i.coins = map[string][]clientlib.Vtxo{} + i.mu.Unlock() + for range 6 { + time.Sleep(time.Second) + e.svc.wakeScan() + synctest.Wait() + } + require.Equal(t, 2, i.requestCount()) + require.Equal(t, 5*time.Second, e.svc.lastFullRefresh.Sub(first)) + require.Equal(t, e.svc.lastFullRefresh.Add(e.svc.pollInterval), e.svc.Status().NextScan) + for _, holding := range e.svc.Status().Holdings { + require.Zero(t, holding.Vtxos) + } + e.svc.Stop() + }) +} + +func TestCacheInvalidationsRespectScanBackoff(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + e := newTestEnv(t) + manyDelegations(t, e, 1) + i := &cacheIndexer{Indexer: e.indexer, coins: map[string][]clientlib.Vtxo{}} + script := e.indexer.vtxos[0].Script + i.coins[script] = e.indexer.vtxos + e.svc.indexer = i + e.svc.pollInterval = 5 * time.Second + e.svc.Start() + synctest.Wait() + first := e.svc.lastFullRefresh + i.mu.Lock() + i.err = ports.ErrThrottled + i.mu.Unlock() + wake := func() { + time.Sleep(time.Second) + e.svc.invalidateScripts(script) + synctest.Wait() + } + wake() + require.Equal(t, 2, i.requestCount()) + require.Equal(t, first.Add(6*time.Second), e.svc.Status().NextScan) + for range 4 { + wake() + } + require.Equal(t, 2, i.requestCount()) + wake() + require.Equal(t, 3, i.requestCount()) + require.Equal(t, first.Add(16*time.Second), e.svc.Status().NextScan) + require.True(t, e.svc.Status().Throttled) + require.Equal(t, first, e.svc.lastFullRefresh) + i.mu.Lock() + i.err = nil + i.mu.Unlock() + for range 9 { + wake() + } + require.Equal(t, 3, i.requestCount()) + wake() + require.Equal(t, 4, i.requestCount()) + require.False(t, e.svc.Status().Throttled) + require.Equal(t, first.Add(16*time.Second), e.svc.lastFullRefresh) + e.svc.Stop() + }) +} + +func TestFailedDiscoveryDoesNotAdvanceFullReconciliation(t *testing.T) { + e := newTestEnv(t) + board := e.boarding(t, e.userKey.PubKey()) + e.deposit(t, board, 3000) + e.explorer.utxoErr = ports.ErrThrottled + previous := time.Now().Add(-time.Hour) + e.svc.lastFullRefresh = previous + require.ErrorIs(t, e.svc.scan(t.Context()), ports.ErrThrottled) + require.Equal(t, previous, e.svc.lastFullRefresh) + require.True(t, e.svc.vtxoDirtyAll) +} diff --git a/internal/core/ports/ark.go b/internal/core/ports/ark.go index 53c59ab..86b14c0 100644 --- a/internal/core/ports/ark.go +++ b/internal/core/ports/ark.go @@ -20,12 +20,12 @@ type Ark interface { FinalizeTx(ctx context.Context, arkTxid string, finalCheckpointTxs []string) error // GetPendingTx lists the offchain txs arkd accepted and did not finalize, spending the coins the proof owns. GetPendingTx(ctx context.Context, proof, message string) ([]clientlib.AcceptedOffchainTx, error) - // GetTransactionsStream streams the transactions arkd accepts until ctx ends or its stop function runs. - GetTransactionsStream(ctx context.Context) (<-chan clientlib.TransactionEvent, func(), error) } // Indexer lists the coins and transactions the server knows. type Indexer interface { + NewTransactionSubscription(ctx context.Context, expressions []string) (string, <-chan clientlib.TransactionEvent, func(), error) + UpdateTransactionSubscription(ctx context.Context, subscriptionID string, expressions []string) error GetVtxos(ctx context.Context, opts ...clientlib.GetVtxosOption) (*clientlib.VtxosResponse, error) GetVirtualTxs(ctx context.Context, txids []string, opts ...clientlib.PageOption) (*clientlib.VirtualTxsResponse, error) // NewSubscription streams what happens to scripts until ctx ends or its stop function runs: either closes the channel. diff --git a/internal/infrastructure/indexer/client.go b/internal/infrastructure/indexer/client.go new file mode 100644 index 0000000..609ba08 --- /dev/null +++ b/internal/infrastructure/indexer/client.go @@ -0,0 +1,195 @@ +package indexer + +import ( + "context" + "fmt" + "slices" + "strings" + "sync" + "time" + + arkv1 "github.com/arkade-os/arkd/api-spec/protobuf/gen/ark/v1" + clientlib "github.com/arkade-os/arkd/pkg/client-lib" + upstream "github.com/arkade-os/arkd/pkg/client-lib/indexer" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" +) + +// Client leaves reconnects to the application, which owns the current subscription IDs. +type Client struct { + clientlib.Indexer + conn *grpc.ClientConn + svc arkv1.IndexerServiceClient + ctx context.Context + cancel context.CancelFunc + closeOnce sync.Once +} + +func NewClient(url string) (*Client, error) { + sdk, err := upstream.NewClient(strings.TrimSuffix(url, "/")) + if err != nil { + return nil, err + } + target, creds := grpcTarget(url) + conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(creds), grpc.WithDisableServiceConfig(), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(20<<20))) + if err != nil { + sdk.Close() + return nil, err + } + ctx, cancel := context.WithCancel(context.Background()) + return &Client{Indexer: sdk, conn: conn, svc: arkv1.NewIndexerServiceClient(conn), ctx: ctx, cancel: cancel}, nil +} + +func (c *Client) Close() { + c.closeOnce.Do(func() { c.cancel(); c.Indexer.Close(); _ = c.conn.Close() }) +} + +func (c *Client) NewTransactionSubscription(ctx context.Context, expressions []string) (string, <-chan clientlib.TransactionEvent, func(), error) { + if len(expressions) == 0 { + return "", nil, nil, fmt.Errorf("missing transaction filter expressions") + } + stream, id, readCtx, stop, err := c.open(ctx, &arkv1.SubscriptionFilter{Expressions: slices.Clone(expressions)}) + if err != nil { + return "", nil, nil, err + } + events := readEvents(readCtx, stream, stop, func(event *arkv1.IndexerSubscriptionEvent) clientlib.TransactionEvent { + return clientlib.TransactionEvent{ArkTx: &clientlib.TxNotification{TxData: clientlib.TxData{Txid: event.GetTxid(), Tx: event.GetTx()}}} + }, func(err error) clientlib.TransactionEvent { return clientlib.TransactionEvent{Err: err} }) + return id, events, stop, nil +} + +func (c *Client) UpdateTransactionSubscription(ctx context.Context, id string, expressions []string) error { + if id == "" { + return fmt.Errorf("missing subscription id to update") + } + if len(expressions) == 0 { + return fmt.Errorf("missing transaction filter expressions") + } + _, err := c.svc.UpdateSubscription(ctx, &arkv1.UpdateSubscriptionRequest{SubscriptionId: id, Filter: &arkv1.SubscriptionFilter{Expressions: slices.Clone(expressions)}}) + return err +} + +func (c *Client) NewSubscription(ctx context.Context, scripts []string) (string, <-chan clientlib.ScriptEvent, func(), error) { + if len(scripts) == 0 { + return "", nil, nil, fmt.Errorf("missing subscription scripts") + } + stream, id, readCtx, stop, err := c.open(ctx, &arkv1.SubscriptionFilter{Scripts: &arkv1.ScriptFilter{Add: slices.Clone(scripts)}}) + if err != nil { + return "", nil, nil, err + } + events := readEvents(readCtx, stream, stop, func(event *arkv1.IndexerSubscriptionEvent) clientlib.ScriptEvent { + checkpoints := map[string]clientlib.TxData{} + for key, tx := range event.GetCheckpointTxs() { + checkpoints[key] = clientlib.TxData{Txid: tx.GetTxid(), Tx: tx.GetTx()} + } + return clientlib.ScriptEvent{Data: &clientlib.ScriptEventData{Txid: event.GetTxid(), Tx: event.GetTx(), Scripts: event.GetScripts(), NewVtxos: vtxos(event.GetNewVtxos()), SpentVtxos: vtxos(event.GetSpentVtxos()), CheckpointTxs: checkpoints}} + }, func(err error) clientlib.ScriptEvent { return clientlib.ScriptEvent{Err: err} }) + return id, events, stop, nil +} + +func (c *Client) UpdateSubscription(ctx context.Context, id string, add, remove []string) error { + if id == "" { + return fmt.Errorf("missing subscription id to update") + } + if len(add)+len(remove) == 0 { + return fmt.Errorf("missing scripts to add or remove") + } + _, err := c.svc.UpdateSubscription(ctx, &arkv1.UpdateSubscriptionRequest{SubscriptionId: id, Filter: &arkv1.SubscriptionFilter{Scripts: &arkv1.ScriptFilter{Add: slices.Clone(add), Remove: slices.Clone(remove)}}}) + return err +} + +func (c *Client) open(ctx context.Context, filter *arkv1.SubscriptionFilter) (arkv1.IndexerService_GetSubscriptionClient, string, context.Context, func(), error) { + readCtx, cancel := context.WithCancel(ctx) + detach := context.AfterFunc(c.ctx, cancel) + stop := func() { cancel(); detach() } + timer := time.AfterFunc(10*time.Second, cancel) + defer timer.Stop() + stream, err := c.svc.GetSubscription(readCtx, &arkv1.GetSubscriptionRequest{Filter: filter}) + if err != nil { + stop() + return nil, "", nil, nil, err + } + for { + response, err := stream.Recv() + if err != nil { + stop() + return nil, "", nil, nil, err + } + if response.GetHeartbeat() != nil { + continue + } + ack := response.GetSubscriptionStarted() + if ack == nil || ack.GetSubscriptionId() == "" { + stop() + return nil, "", nil, nil, fmt.Errorf("missing subscription acknowledgement") + } + return stream, ack.GetSubscriptionId(), readCtx, stop, nil + } +} + +func readEvents[T any](ctx context.Context, stream arkv1.IndexerService_GetSubscriptionClient, stop func(), convert func(*arkv1.IndexerSubscriptionEvent) T, failure func(error) T) <-chan T { + events := make(chan T, 1) + go func() { + defer close(events) + defer stop() + for { + response, err := stream.Recv() + if ctx.Err() != nil { + return + } + var event T + if err != nil { + event = failure(err) + } else if data := response.GetEvent(); data != nil { + event = convert(data) + } else { + continue + } + select { + case <-ctx.Done(): + return + case events <- event: + } + if err != nil { + return + } + } + }() + return events +} + +func vtxos(items []*arkv1.IndexerVtxo) []clientlib.Vtxo { + out := make([]clientlib.Vtxo, 0, len(items)) + for _, v := range items { + var assets []clientlib.Asset + for _, a := range v.GetAssets() { + if a != nil { + assets = append(assets, clientlib.Asset{AssetId: a.GetAssetId(), Amount: a.GetAmount()}) + } + } + out = append(out, clientlib.Vtxo{ + Outpoint: clientlib.Outpoint{Txid: v.GetOutpoint().GetTxid(), VOut: v.GetOutpoint().GetVout()}, + Script: v.GetScript(), CommitmentTxids: v.GetCommitmentTxids(), Amount: v.GetAmount(), + CreatedAt: time.Unix(v.GetCreatedAt(), 0), ExpiresAt: time.Unix(v.GetExpiresAt(), 0), + Preconfirmed: v.GetIsPreconfirmed(), Swept: v.GetIsSwept(), Spent: v.GetIsSpent(), + Unrolled: v.GetIsUnrolled(), SpentBy: v.GetSpentBy(), SettledBy: v.GetSettledBy(), + ArkTxid: v.GetArkTxid(), Assets: assets, + }) + } + return out +} + +func grpcTarget(url string) (string, credentials.TransportCredentials) { + creds := insecure.NewCredentials() + port := "80" + if strings.HasPrefix(url, "https://") { + creds = credentials.NewTLS(nil) + port = "443" + } + target := strings.TrimSuffix(strings.TrimPrefix(strings.TrimPrefix(url, "https://"), "http://"), "/") + if !strings.Contains(target, ":") { + target += ":" + port + } + return target, creds +} diff --git a/internal/infrastructure/indexer/client_test.go b/internal/infrastructure/indexer/client_test.go new file mode 100644 index 0000000..b1dde47 --- /dev/null +++ b/internal/infrastructure/indexer/client_test.go @@ -0,0 +1,331 @@ +package indexer + +import ( + "context" + "net" + "sync/atomic" + "testing" + "time" + + arkv1 "github.com/arkade-os/arkd/api-spec/protobuf/gen/ark/v1" + clientlib "github.com/arkade-os/arkd/pkg/client-lib" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type subscriptions interface { + clientlib.Indexer + NewTransactionSubscription(context.Context, []string) (string, <-chan clientlib.TransactionEvent, func(), error) + UpdateTransactionSubscription(context.Context, string, []string) error +} + +type subscriptionServer struct { + arkv1.UnimplementedIndexerServiceServer + subscribe func(*arkv1.GetSubscriptionRequest, arkv1.IndexerService_GetSubscriptionServer) error + update func(*arkv1.UpdateSubscriptionRequest) error + calls atomic.Int32 +} + +func (s *subscriptionServer) GetSubscription(req *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + s.calls.Add(1) + return s.subscribe(req, stream) +} +func (s *subscriptionServer) UpdateSubscription(_ context.Context, req *arkv1.UpdateSubscriptionRequest) (*arkv1.UpdateSubscriptionResponse, error) { + return &arkv1.UpdateSubscriptionResponse{}, s.update(req) +} +func testClient(t *testing.T, s *subscriptionServer) subscriptions { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + server := grpc.NewServer() + arkv1.RegisterIndexerServiceServer(server, s) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + client, err := NewClient(listener.Addr().String()) + require.NoError(t, err) + t.Cleanup(client.Close) + require.Implements(t, (*subscriptions)(nil), client) + return client +} +func started(stream arkv1.IndexerService_GetSubscriptionServer, id string) error { + return stream.Send(&arkv1.GetSubscriptionResponse{Data: &arkv1.GetSubscriptionResponse_SubscriptionStarted{SubscriptionStarted: &arkv1.SubscriptionStartedEvent{SubscriptionId: id}}}) +} +func txEvent(txid, tx string) *arkv1.GetSubscriptionResponse { + return &arkv1.GetSubscriptionResponse{Data: &arkv1.GetSubscriptionResponse_Event{Event: &arkv1.IndexerSubscriptionEvent{Txid: txid, Tx: tx}}} +} +func TestTransactionFilterAndServerFailure(t *testing.T) { + requests := make(chan *arkv1.GetSubscriptionRequest, 1) + server := &subscriptionServer{subscribe: func(req *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + requests <- req + if err := started(stream, "transaction-id"); err != nil { + return err + } + if err := stream.Send(&arkv1.GetSubscriptionResponse{Data: &arkv1.GetSubscriptionResponse_Heartbeat{Heartbeat: &arkv1.IndexerHeartbeat{}}}); err != nil { + return err + } + if err := stream.Send(txEvent("matched", "full-psbt")); err != nil { + return err + } + return status.Error(codes.Unavailable, "lost server") + }} + client := testClient(t, server) + expressions := []string{"tx.extensions.exists(e, e.type == 4)"} + _, events, stop, err := client.NewTransactionSubscription(t.Context(), expressions) + require.NoError(t, err) + defer stop() + req := <-requests + require.Empty(t, req.SubscriptionId) + require.Equal(t, expressions, req.Filter.Expressions) + require.Nil(t, req.Filter.Scripts) + first := <-events + require.NoError(t, first.Err) + require.Equal(t, "full-psbt", first.ArkTx.Tx) + require.Equal(t, "matched", first.ArkTx.Txid) + next := <-events + require.Equal(t, codes.Unavailable, status.Code(next.Err)) + _, ok := <-events + require.False(t, ok) + require.EqualValues(t, 1, server.calls.Load()) +} +func TestStartupRejection(t *testing.T) { + for _, code := range []codes.Code{codes.Unimplemented, codes.InvalidArgument} { + t.Run(code.String(), func(t *testing.T) { + server := &subscriptionServer{subscribe: func(*arkv1.GetSubscriptionRequest, arkv1.IndexerService_GetSubscriptionServer) error { + return status.Error(code, "filter rejected") + }} + client := testClient(t, server) + _, events, stop, err := client.NewTransactionSubscription(t.Context(), []string{"filter"}) + require.Equal(t, code, status.Code(err)) + require.Nil(t, events) + require.Nil(t, stop) + require.EqualValues(t, 1, server.calls.Load()) + }) + } +} +func TestRequiresSubscriptionAcknowledgement(t *testing.T) { + server := &subscriptionServer{subscribe: func(_ *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + return stream.Send(txEvent("unfiltered", "psbt")) + }} + client := testClient(t, server) + _, events, stop, err := client.NewTransactionSubscription(t.Context(), []string{"filter"}) + require.ErrorContains(t, err, "acknowledgement") + require.Nil(t, events) + require.Nil(t, stop) +} +func TestCancellationUnblocksBackpressure(t *testing.T) { + for _, closeClient := range []bool{false, true} { + t.Run(map[bool]string{false: "subscription", true: "client"}[closeClient], func(t *testing.T) { + sent := make(chan struct{}) + ended := make(chan struct{}) + server := &subscriptionServer{subscribe: func(_ *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + defer close(ended) + if err := started(stream, "backpressure"); err != nil { + return err + } + for range 20 { + if err := stream.Send(txEvent("match", "psbt")); err != nil { + return err + } + } + close(sent) + <-stream.Context().Done() + return stream.Context().Err() + }} + client := testClient(t, server) + _, events, stop, err := client.NewTransactionSubscription(t.Context(), []string{"filter"}) + require.NoError(t, err) + <-sent + require.LessOrEqual(t, cap(events), 16) + if closeClient { + client.Close() + } else { + stop() + } + drained := make(chan struct{}) + go func() { + for range events { + } + close(drained) + }() + select { + case <-drained: + case <-time.After(time.Second): + t.Fatal("event reader leaked") + } + select { + case <-ended: + case <-time.After(time.Second): + t.Fatal("server stream leaked") + } + stop() + client.Close() + }) + } +} +func TestScriptSubscriptionUsesAcknowledgedID(t *testing.T) { + requests := make(chan *arkv1.GetSubscriptionRequest, 1) + updates := make(chan *arkv1.UpdateSubscriptionRequest, 1) + server := &subscriptionServer{ + subscribe: func(req *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + requests <- req + if err := started(stream, "scripts-id"); err != nil { + return err + } + ev := txEvent("funding", "psbt") + ev.GetEvent().Scripts = []string{"5120aa"} + ev.GetEvent().NewVtxos = []*arkv1.IndexerVtxo{{Amount: 42, Script: "5120aa"}} + if err := stream.Send(ev); err != nil { + return err + } + <-stream.Context().Done() + return stream.Context().Err() + }, update: func(req *arkv1.UpdateSubscriptionRequest) error { updates <- req; return nil }} + client := testClient(t, server) + id, events, stop, err := client.NewSubscription(t.Context(), []string{"5120aa"}) + require.NoError(t, err) + defer stop() + require.Equal(t, "scripts-id", id) + req := <-requests + require.Equal(t, []string{"5120aa"}, req.Filter.Scripts.Add) + require.Empty(t, req.Filter.Expressions) + ev := <-events + require.NoError(t, ev.Err) + require.Equal(t, "psbt", ev.Data.Tx) + require.EqualValues(t, 42, ev.Data.NewVtxos[0].Amount) + require.NoError(t, client.UpdateSubscription(t.Context(), id, []string{"5120bb"}, []string{"5120aa"})) + update := <-updates + require.Equal(t, id, update.SubscriptionId) + require.Equal(t, []string{"5120bb"}, update.Filter.Scripts.Add) + require.Equal(t, []string{"5120aa"}, update.Filter.Scripts.Remove) + stop() + _, ok := <-events + require.False(t, ok) + require.EqualValues(t, 1, server.calls.Load()) +} + +func TestEmptyFiltersNeverOpenASubscription(t *testing.T) { + server := &subscriptionServer{subscribe: func(*arkv1.GetSubscriptionRequest, arkv1.IndexerService_GetSubscriptionServer) error { + return status.Error(codes.Internal, "unfiltered stream opened") + }} + client := testClient(t, server) + _, events, stop, err := client.NewTransactionSubscription(t.Context(), nil) + require.ErrorContains(t, err, "expressions") + require.Nil(t, events) + require.Nil(t, stop) + id, scripts, stop, err := client.NewSubscription(t.Context(), nil) + require.ErrorContains(t, err, "scripts") + require.Empty(t, id) + require.Nil(t, scripts) + require.Nil(t, stop) + require.Zero(t, server.calls.Load()) +} + +func TestParentCancellationDuringStartup(t *testing.T) { + opened := make(chan struct{}) + ended := make(chan struct{}) + server := &subscriptionServer{subscribe: func(_ *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + close(opened) + defer close(ended) + <-stream.Context().Done() + return stream.Context().Err() + }} + client := testClient(t, server) + ctx, cancel := context.WithCancel(t.Context()) + failed := make(chan error, 1) + go func() { _, _, _, err := client.NewTransactionSubscription(ctx, []string{"filter"}); failed <- err }() + <-opened + cancel() + select { + case err := <-failed: + require.Equal(t, codes.Canceled, status.Code(err)) + case <-time.After(time.Second): + t.Fatal("startup ignores cancellation") + } + select { + case <-ended: + case <-time.After(time.Second): + t.Fatal("startup stream leaked") + } +} + +func TestScriptFailureClosesWithoutReplacingID(t *testing.T) { + server := &subscriptionServer{subscribe: func(_ *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + if err := started(stream, "original-id"); err != nil { + return err + } + return status.Error(codes.Unavailable, "disconnected") + }, update: func(*arkv1.UpdateSubscriptionRequest) error { return status.Error(codes.NotFound, "expired id") }} + client := testClient(t, server) + id, events, stop, err := client.NewSubscription(t.Context(), []string{"script"}) + require.NoError(t, err) + defer stop() + require.Equal(t, "original-id", id) + event := <-events + require.Equal(t, codes.Unavailable, status.Code(event.Err)) + _, ok := <-events + require.False(t, ok) + require.Equal(t, codes.NotFound, status.Code(client.UpdateSubscription(t.Context(), id, []string{"other"}, nil))) + require.EqualValues(t, 1, server.calls.Load()) +} + +func TestGRPCTarget(t *testing.T) { + for _, tc := range []struct{ url, target, security string }{ + {"localhost:7070", "localhost:7070", "insecure"}, {"http://localhost/", "localhost:80", "insecure"}, {"https://ark.example/", "ark.example:443", "tls"}, {"https://ark.example:7070", "ark.example:7070", "tls"}, + } { + target, creds := grpcTarget(tc.url) + require.Equal(t, tc.target, target) + require.Equal(t, tc.security, creds.Info().SecurityProtocol) + } +} +func TestTransactionFilterUpdateKeepsStream(t *testing.T) { + emitted := make(chan string, 2) + updates := make(chan *arkv1.UpdateSubscriptionRequest, 2) + server := &subscriptionServer{ + subscribe: func(_ *arkv1.GetSubscriptionRequest, stream arkv1.IndexerService_GetSubscriptionServer) error { + if err := started(stream, "stable-id"); err != nil { + return err + } + for { + select { + case <-stream.Context().Done(): + return stream.Context().Err() + case tx := <-emitted: + if err := stream.Send(txEvent(tx, "psbt")); err != nil { + return err + } + } + } + }, + update: func(req *arkv1.UpdateSubscriptionRequest) error { + updates <- req + if req.Filter.Expressions[0] == "invalid" { + return status.Error(codes.InvalidArgument, "invalid expression") + } + return nil + }, + } + client := testClient(t, server) + id, events, stop, err := client.NewTransactionSubscription(t.Context(), []string{"initial"}) + require.NoError(t, err) + defer stop() + require.Equal(t, "stable-id", id) + require.NoError(t, client.UpdateTransactionSubscription(t.Context(), id, []string{"replacement"})) + req := <-updates + require.Equal(t, id, req.SubscriptionId) + require.Equal(t, []string{"replacement"}, req.Filter.Expressions) + require.Nil(t, req.Filter.Scripts) + emitted <- "after update" + require.Equal(t, "after update", (<-events).ArkTx.Txid) + require.Equal(t, codes.InvalidArgument, status.Code(client.UpdateTransactionSubscription(t.Context(), id, []string{"invalid"}))) + req = <-updates + require.Equal(t, id, req.SubscriptionId) + require.Nil(t, req.Filter.Scripts) + emitted <- "after rejection" + require.Equal(t, "after rejection", (<-events).ArkTx.Txid) + require.EqualValues(t, 1, server.calls.Load()) + require.ErrorContains(t, client.UpdateTransactionSubscription(t.Context(), "", []string{"replacement"}), "id") + require.ErrorContains(t, client.UpdateTransactionSubscription(t.Context(), id, nil), "expressions") + require.Empty(t, updates) +} diff --git a/pkg/template/contract.go b/pkg/template/contract.go index e969dc6..f294cee 100644 --- a/pkg/template/contract.go +++ b/pkg/template/contract.go @@ -38,6 +38,18 @@ func buildTree(def *Definition, args map[string]value, keys Keys, taptree []byte } return v, nil } + // covenants and leaves both see a pubkey as the x-only key taproot uses + keyField := func(name string) (value, error) { + v, err := field(name) + if err != nil || v.typ != "pubkey" { + return v, err + } + pk, err := btcec.ParsePubKey(v.raw) + if err != nil { + return value{}, err + } + return value{"bytes32", schnorr.SerializePubKey(pk)}, nil + } delegate := func() (*btcec.PublicKey, error) { if keys.Delegate == nil { @@ -48,7 +60,7 @@ func buildTree(def *Definition, args map[string]value, keys Keys, taptree []byte // intent messages spell cosigner keys in hex covenantLookup := func(name string) (value, error) { if name != "DELEGATE_KEY" { - return field(name) + return keyField(name) } pk, err := delegate() if err != nil { @@ -101,18 +113,7 @@ func buildTree(def *Definition, args map[string]value, keys Keys, taptree []byte } return tweak(pk, parts[2]) } - v, err := field(name) - if err != nil { - return value{}, err - } - if v.typ == "pubkey" { - pk, err := btcec.ParsePubKey(v.raw) - if err != nil { - return value{}, err - } - return value{"bytes32", schnorr.SerializePubKey(pk)}, nil - } - return v, nil + return keyField(name) } var leaves []txscript.TapLeaf diff --git a/pkg/template/contract_test.go b/pkg/template/contract_test.go index bb69d53..a8e9400 100644 --- a/pkg/template/contract_test.go +++ b/pkg/template/contract_test.go @@ -63,6 +63,21 @@ func TestBuildTreeShapes(t *testing.T) { } } +// A covenant compares a key against a taproot witness program, so it needs the +// x-only bytes a leaf gets; arkadec's own e2e binds pubkeys the same way. +func TestBuildCovenantPushesXOnlyPubkey(t *testing.T) { + owner := testKey(t, 2) + def := parseDef(t, `{"contractName":"C","constructorInputs":[{"name":"owner","type":"pubkey"}],"functions":[ + {"name":"pay","arkade":{"inputs":[],"asm":["","OP_DROP","OP_1"]},"leaves":[{"name":"pay","asm":["","OP_CHECKSIGVERIFY","","OP_CHECKSIG"],"witness":[]}]}]}`) + c, err := build(def, map[string]value{"owner": {"pubkey", owner.SerializeCompressed()}}, Keys{Server: testKey(t, 1), Emulator: testKey(t, 3)}) + require.NoError(t, err) + + want, err := txscript.NewScriptBuilder(). + AddData(schnorr.SerializePubKey(owner)).AddOp(txscript.OP_DROP).AddOp(txscript.OP_1).Script() + require.NoError(t, err) + require.Equal(t, want, c.covenants["pay"]) +} + func TestBuildRejects(t *testing.T) { keys := Keys{Server: testKey(t, 1), Emulator: testKey(t, 3)} def := parseDef(t, `{"contractName":"C","constructorInputs":[],"functions":[ diff --git a/pkg/template/fixtures_test.go b/pkg/template/fixtures_test.go index 126d703..d016f50 100644 --- a/pkg/template/fixtures_test.go +++ b/pkg/template/fixtures_test.go @@ -33,7 +33,7 @@ func TestFixtures(t *testing.T) { var ids map[string]string require.NoError(t, json.Unmarshal(fixture(t, "ids.json"), &ids)) resolve := artifacts(t) - for _, name := range []string{"renewal.json", "boarding.json", "counter.json", "vhtlc_claim.json", "vhtlc_refund.json", "noninteractiveclaim.json", "onchain_release.json", "counter_seed.json", "minimal.json"} { + for _, name := range []string{"renewal.json", "boarding.json", "counter.json", "vhtlc_claim.json", "vhtlc_refund.json", "noninteractiveclaim.json", "offline_receive.json", "onchain_release.json", "counter_seed.json", "minimal.json"} { t.Run(name, func(t *testing.T) { tmpl, err := Parse(t.Context(), fixture(t, name), resolve) require.NoError(t, err) diff --git a/pkg/template/go.mod b/pkg/template/go.mod index 7d46e29..8324052 100644 --- a/pkg/template/go.mod +++ b/pkg/template/go.mod @@ -10,12 +10,13 @@ require ( github.com/btcsuite/btcd/psbt/v2 v2.0.0 github.com/btcsuite/btcd/txscript/v2 v2.0.0 github.com/btcsuite/btcd/wire/v2 v2.0.1 + github.com/google/cel-go v0.30.0 github.com/stretchr/testify v1.12.1 ) require ( cel.dev/expr v0.25.2 // indirect - github.com/antlr4-go/antlr/v4 v4.13.0 // indirect + github.com/antlr4-go/antlr/v4 v4.13.1 // indirect github.com/arkade-os/arkd/pkg/errors v0.0.0-20260901090427-f863e4847193 // indirect github.com/bits-and-blooms/bitset v1.20.0 // indirect github.com/btcsuite/btcd v0.26.2 // indirect @@ -26,10 +27,8 @@ require ( github.com/consensys/gnark-crypto v0.19.2 // indirect github.com/decred/dcrd/crypto/blake256 v1.1.0 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect - github.com/google/cel-go v0.26.1 // indirect github.com/kcalvinalvin/anet v0.0.0-20251112173137-d8ddc1f6dbee // indirect github.com/sirupsen/logrus v1.9.3 // indirect - github.com/stoewer/go-strcase v1.2.0 // indirect github.com/tidwall/gjson v1.19.0 // indirect github.com/tidwall/match v1.1.1 // indirect github.com/tidwall/pretty v1.2.0 // indirect @@ -37,6 +36,7 @@ require ( golang.org/x/crypto v0.57.0 // indirect golang.org/x/exp v0.0.0-20250811191247-51f88131bc50 // indirect golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.42.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/grpc v1.83.2 // indirect diff --git a/pkg/template/go.sum b/pkg/template/go.sum index c38a58a..71f54ec 100644 --- a/pkg/template/go.sum +++ b/pkg/template/go.sum @@ -2,8 +2,8 @@ cel.dev/expr v0.25.2 h1:K6j46C81hXtZQfuX60cVWQFBJahKSE2gfRbNuvr5bFs= cel.dev/expr v0.25.2/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= -github.com/antlr4-go/antlr/v4 v4.13.0 h1:lxCg3LAv+EUK6t1i0y1V6/SLeUi0eKEKdhQAlS8TVTI= -github.com/antlr4-go/antlr/v4 v4.13.0/go.mod h1:pfChB/xh/Unjila75QW7+VU4TSnWnnk9UTnmpPaOR2g= +github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= +github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw= github.com/arkade-os/arkd/pkg/ark-lib v0.8.1-0.20260901090427-f863e4847193 h1:MKvnk+x98t4K7HnHjWhsi56y+jSIGH6KNtwp7H27CkI= github.com/arkade-os/arkd/pkg/ark-lib v0.8.1-0.20260901090427-f863e4847193/go.mod h1:PYUq0KIVEBp6nmPAQP+A6hFtXDixjvpYGrBWsMaRCLY= github.com/arkade-os/arkd/pkg/errors v0.0.0-20260901090427-f863e4847193 h1:8Mk7Lhh5NIpZlz196sjIyBcNuVIAs5ttngi8/rWzKUY= @@ -44,8 +44,8 @@ github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 h1:NMZiJj8QnKe1LgsbDayM4UoHwbvw github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40= github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs= github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= -github.com/google/cel-go v0.26.1 h1:iPbVVEdkhTX++hpe3lzSk7D3G3QSYqLGoHOcEio+UXQ= -github.com/google/cel-go v0.26.1/go.mod h1:A9O8OU9rdvrK5MQyrqfIxo1a0u4g3sF8KB6PUIaryMM= +github.com/google/cel-go v0.30.0 h1:ll54AkzKunWkBn9wSoiUXbFZXYZTkdJGNXTBXUoolGo= +github.com/google/cel-go v0.30.0/go.mod h1:X0bD6iVNR8pkROSOoHVdgTkzmRcosof7WQqCD6wcMc8= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/kcalvinalvin/anet v0.0.0-20251112173137-d8ddc1f6dbee h1:FPP9HDkBbPyniu+u7FHZg+kKFX1WW0gxOGteJ0h3AJk= @@ -55,10 +55,7 @@ github.com/leanovate/gopter v0.2.11/go.mod h1:aK3tzZP/C+p1m3SPRE4SYZFGP7jjkuSI4f github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= -github.com/stoewer/go-strcase v1.2.0 h1:Z2iHWqGXH00XYgqDmNgQbIBxf3wrNq0F3feEy0ainaU= -github.com/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag1KpM8ahLw8= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= -github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= @@ -90,9 +87,6 @@ google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsok google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= diff --git a/pkg/template/instance.go b/pkg/template/instance.go index a00b7cb..e001341 100644 --- a/pkg/template/instance.go +++ b/pkg/template/instance.go @@ -3,6 +3,8 @@ package template import ( "bytes" "context" + "encoding/hex" + "encoding/json" "fmt" "math" "slices" @@ -362,6 +364,20 @@ func (i *Instance) lookup(n string, d *draft) (value, error) { return value{}, fmt.Errorf("secret %q: %w", n, err) } p, err := i.decrypt(ct) + if err == nil && s.field != "" { + var envelope struct { + Format string `json:"format"` + TemplateID string `json:"template_id"` + Values map[string]string `json:"values"` + } + if e := json.Unmarshal(p, &envelope); e != nil || envelope.Format != "delegatee-secrets/v1" || envelope.TemplateID != i.tmpl.id { + err = fmt.Errorf("invalid encrypted secret envelope") + } else if raw, ok := envelope.Values[s.field]; !ok { + err = fmt.Errorf("encrypted secret field is missing") + } else if p, err = hex.DecodeString(raw); err != nil || hex.EncodeToString(p) != raw { + err = fmt.Errorf("encrypted secret field is not hex") + } + } if err == nil { err = canonicalBytes(s.typ, p) } diff --git a/pkg/template/template.go b/pkg/template/template.go index 406acbf..cdd8b35 100644 --- a/pkg/template/template.go +++ b/pkg/template/template.go @@ -119,6 +119,7 @@ type secret struct { typ string from int ciphertext byteTemplate + field string } type fee struct { @@ -277,7 +278,7 @@ func (p *parser) declarations(m map[string]any) (map[string]any, error) { if _, dup := p.t.variables[n]; dup { return nil, bad("%q is both a variable and a secret", n) } - sm, err := object(ErrInvalidTemplate, v, "secret", []string{"type", "from", "ciphertext"}) + sm, err := object(ErrInvalidTemplate, v, "secret", []string{"type", "from", "ciphertext"}, "field") if err != nil { return nil, err } @@ -285,7 +286,15 @@ func (p *parser) declarations(m map[string]any) (map[string]any, error) { if !slices.Contains(scalarTypes, typ) { return nil, bad("secret %q has unsupported type %v", n, sm["type"]) } - p.t.secrets[n] = secret{typ: typ} + field := "" + if raw, ok := sm["field"]; ok { + var valid bool + field, valid = raw.(string) + if !valid || ident(field, "secret field") != nil { + return nil, bad("secret %q has an invalid field", n) + } + } + p.t.secrets[n] = secret{typ: typ, field: field} } if len(vars)+len(secrets) > maxSlots { return nil, bad("more than %d variables and secrets", maxSlots) diff --git a/pkg/template/testdata/ids.json b/pkg/template/testdata/ids.json index efce18b..1a763c3 100644 --- a/pkg/template/testdata/ids.json +++ b/pkg/template/testdata/ids.json @@ -7,5 +7,6 @@ "onchain_release.json": "49886955464339b989f4a667b060652cbfcb4416eec5cd356f68521ffbbf8205", "counter_seed.json": "9b5c01dc328cb4b694a0177c40d22687f658fc6828ca9296d75fa15fdb852ca0", "noninteractiveclaim.json": "4493579d4cdb14a5809b55c55b7d228a4d8f7fc479c1424b286d4d5234382eee", + "offline_receive.json": "fe39de17a0f608b710d9290b7533ec48204493a3e7cf726e03098b996eea38a2", "minimal.json": "3f5f97fb405af0cfadfdc5b9729c668a66b684916372099af13f6bcb45265477" } diff --git a/pkg/template/trigger_filter.go b/pkg/template/trigger_filter.go new file mode 100644 index 0000000..9c4ea45 --- /dev/null +++ b/pkg/template/trigger_filter.go @@ -0,0 +1,40 @@ +package template + +import ( + "encoding/binary" + "encoding/hex" + "fmt" + "math" + "strconv" + "strings" +) + +// Filter is a conservative Arkd CEL prefilter; Match remains authoritative. +func (tr *Trigger) Filter(keys TriggerKeys) string { + clauses := []string{fmt.Sprintf("has(tx.extension) && hasPacket(tx.extension, %d)", tr.packet)} + for _, r := range tr.require { + var alternatives []string + for _, value := range r.values(keys) { + if tr.raw { + alternatives = append(alternatives, fmt.Sprintf("tx.extension[%d] == '%s'", tr.packet, hex.EncodeToString(value))) + continue + } + if len(value) > math.MaxUint16 { + continue + } + tag, _ := strconv.Atoi(strings.TrimPrefix(r.source, "tlv:")) + record := binary.BigEndian.AppendUint16([]byte{byte(tag)}, uint16(len(value))) + record = append(record, value...) + alternatives = append(alternatives, fmt.Sprintf("tx.extension[%d].contains('%s')", tr.packet, hex.EncodeToString(record))) + } + if len(alternatives) == 0 { + return "false" + } + clauses = append(clauses, "("+strings.Join(alternatives, " || ")+")") + } + filter := strings.Join(clauses, " && ") + if len(filter) > 16<<10 { + return clauses[0] + } + return filter +} diff --git a/pkg/template/trigger_filter_test.go b/pkg/template/trigger_filter_test.go new file mode 100644 index 0000000..a6b6baa --- /dev/null +++ b/pkg/template/trigger_filter_test.go @@ -0,0 +1,165 @@ +package template + +import ( + "bytes" + "encoding/hex" + "reflect" + "testing" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/google/cel-go/cel" + "github.com/google/cel-go/common/types" + "github.com/google/cel-go/common/types/ref" + "github.com/google/cel-go/common/types/traits" + "github.com/google/cel-go/ext" + "github.com/stretchr/testify/require" +) + +type FilterTestTx struct { + Extension map[int64]string `json:"extension,omitempty"` +} + +func filterEvaluator(t *testing.T, tr *Trigger, keys TriggerKeys) func(map[int64]string) bool { + t.Helper() + env, err := cel.NewEnv( + ext.NativeTypes(reflect.TypeFor[FilterTestTx](), ext.ParseStructTag("json")), + cel.Variable("tx", cel.ObjectType("template.FilterTestTx")), + cel.Function("hasPacket", cel.Overload("hasPacket_map_int_string_int_bool", + []*cel.Type{cel.MapType(cel.IntType, cel.StringType), cel.IntType}, cel.BoolType, + cel.BinaryBinding(func(extension, packet ref.Val) ref.Val { + return extension.(traits.Mapper).Contains(packet) + }))), + ) + require.NoError(t, err) + ast, issues := env.Compile(tr.Filter(keys)) + require.NoError(t, issues.Err()) + require.Equal(t, cel.BoolType, ast.OutputType()) + program, err := env.Program(ast) + require.NoError(t, err) + return func(extension map[int64]string) bool { + result, _, err := program.Eval(map[string]any{"tx": FilterTestTx{Extension: extension}}) + require.NoError(t, err) + return result == types.True + } +} + +func TestTriggerFilterRequirements(t *testing.T) { + current, retained, foreign := testKey(t, 1), testKey(t, 2), testKey(t, 3) + keys := TriggerKeys{Encryption: []*btcec.PublicKey{current, nil, retained}, Keys: testKeys(t)} + marker := []byte("delegatee-secrets/v1") + tr := &Trigger{packet: 4, require: []requirement{ + {source: "tlv:3", key: "ENCRYPTION_KEY"}, + {source: "tlv:9", lit: marker}, + }} + eval := filterEvaluator(t, tr, keys) + for _, key := range []*btcec.PublicKey{current, retained} { + body := tlv(9, marker, 3, key.SerializeCompressed()) + require.True(t, eval(map[int64]string{4: hex.EncodeToString(body)})) + ptx, _ := streamed(t, body) + _, err := tr.Match(ptx, keys) + require.NoError(t, err) + } + for name, body := range map[string][]byte{ + "foreign key": tlv(9, marker, 3, foreign.SerializeCompressed()), + "no key": tlv(9, marker), + "no marker": tlv(3, current.SerializeCompressed()), + "wrong tag": tlv(9, marker, 4, current.SerializeCompressed()), + "wrong length": tlv(9, marker, 3, append(current.SerializeCompressed(), 0)), + } { + t.Run(name, func(t *testing.T) { + require.False(t, eval(map[int64]string{4: hex.EncodeToString(body)})) + }) + } + require.False(t, eval(nil)) + require.False(t, eval(map[int64]string{5: hex.EncodeToString(tlv(9, marker, 3, current.SerializeCompressed()))})) + require.Equal(t, "false", tr.Filter(TriggerKeys{})) + require.False(t, filterEvaluator(t, tr, TriggerKeys{})(nil)) +} + +func TestTriggerFilterLiteralBoundaries(t *testing.T) { + for _, size := range []int{0, 1, 255, 256} { + value := bytes.Repeat([]byte{0xab}, size) + tr := &Trigger{packet: 255, require: []requirement{{source: "tlv:7", lit: value}}} + eval := filterEvaluator(t, tr, TriggerKeys{}) + require.True(t, eval(map[int64]string{255: hex.EncodeToString(tlv(7, value))})) + require.False(t, eval(map[int64]string{255: hex.EncodeToString(tlv(8, value))})) + require.False(t, eval(map[int64]string{255: hex.EncodeToString(tlv(7, append(value, 0)))})) + } + tr := &Trigger{packet: 4, require: []requirement{{source: "tlv:1", lit: make([]byte, 65536)}}} + require.Equal(t, "false", tr.Filter(TriggerKeys{})) +} + +func TestTriggerFilterOversizedFallsBack(t *testing.T) { + for _, raw := range []bool{false, true} { + value := bytes.Repeat([]byte{0xab}, 65535) + source := "tlv:7" + body := tlv(7, value) + if raw { + source, body = "raw", value + } + tr := &Trigger{packet: 4, raw: raw, require: []requirement{{source: source, lit: value}}} + require.Equal(t, "has(tx.extension) && hasPacket(tx.extension, 4)", tr.Filter(TriggerKeys{})) + eval := filterEvaluator(t, tr, TriggerKeys{}) + require.True(t, eval(map[int64]string{4: hex.EncodeToString(body)})) + require.False(t, eval(map[int64]string{5: hex.EncodeToString(body)})) + ptx, _ := streamed(t, body) + _, err := tr.Match(ptx, TriggerKeys{}) + require.NoError(t, err) + tr.require = append(tr.require, requirement{source: source, key: "ENCRYPTION_KEY"}) + require.Equal(t, "false", tr.Filter(TriggerKeys{})) + } + tr := &Trigger{packet: 4} + for range 100 { + tr.require = append(tr.require, requirement{source: "tlv:7", lit: bytes.Repeat([]byte{1}, 100)}) + } + require.Equal(t, "has(tx.extension) && hasPacket(tx.extension, 4)", tr.Filter(TriggerKeys{})) +} + +func TestTriggerFilterRawAndPacketOnly(t *testing.T) { + for _, key := range []string{"SERVER_KEY", "EMULATOR_KEY", "DELEGATE_KEY"} { + tr := &Trigger{packet: 4, raw: true, require: []requirement{{source: "raw", key: key}}} + keys := TriggerKeys{Keys: testKeys(t)} + body := tr.require[0].values(keys)[0] + eval := filterEvaluator(t, tr, keys) + require.True(t, eval(map[int64]string{4: hex.EncodeToString(body)})) + require.False(t, eval(map[int64]string{4: hex.EncodeToString(append(body, 0))})) + require.Equal(t, "false", tr.Filter(TriggerKeys{})) + } + tr := &Trigger{packet: 4, raw: true, require: []requirement{{source: "raw", lit: []byte{0xab}}}} + eval := filterEvaluator(t, tr, TriggerKeys{}) + require.True(t, eval(map[int64]string{4: "ab"})) + require.False(t, eval(map[int64]string{4: "00ab"})) + require.False(t, eval(map[int64]string{4: "ab00"})) + tr.require = nil + eval = filterEvaluator(t, tr, TriggerKeys{}) + require.True(t, eval(map[int64]string{4: ""})) + require.False(t, eval(map[int64]string{5: "ab"})) + require.False(t, eval(nil)) +} + +func TestTriggerFilterMatchRemainsAuthoritative(t *testing.T) { + keys := TriggerKeys{Keys: testKeys(t)} + tr := parseDoc(t, triggerDoc(t, baseTrigger())).Trigger() + eval := filterEvaluator(t, tr, keys) + key := keys.Server.SerializeCompressed() + fields := tlv(1, []byte("hi"), 2, pushScript(t, bytes.Repeat([]byte{7}, 32))) + for name, records := range map[string][]byte{ + "embedded record": tlv(9, tlv(3, key)), + "duplicate override": tlv(3, key, 3, testKey(t, 7).SerializeCompressed()), + } { + t.Run(name, func(t *testing.T) { + body := append(bytes.Clone(fields), records...) + require.True(t, eval(map[int64]string{4: hex.EncodeToString(body)})) + ptx, _ := streamed(t, body) + _, err := tr.Match(ptx, keys) + require.ErrorIs(t, err, ErrNoMatch) + }) + } + for _, records := range [][]byte{tlv(3, key), tlv(3, []byte{0}, 3, key)} { + body := append(bytes.Clone(fields), records...) + ptx, _ := streamed(t, body) + _, err := tr.Match(ptx, keys) + require.NoError(t, err) + require.True(t, eval(map[int64]string{4: hex.EncodeToString(body)})) + } +} diff --git a/templates/offline_receive.json b/templates/offline_receive.json new file mode 100644 index 0000000..88df571 --- /dev/null +++ b/templates/offline_receive.json @@ -0,0 +1,169 @@ +{ + "format": "delegateed-template/v1", + "type": "offchain", + "variables": { + "taptree": "bytes", + "ciphertext": "bytes", + "receiver_program": "bytes32" + }, + "secrets": { + "preimage": { + "field": "preimage", + "type": "bytes32", + "ciphertext": "", + "from": "htlc" + }, + "minimum_value": { + "field": "minimum_value", + "type": "int", + "ciphertext": "", + "from": "htlc" + }, + "receiver_commitment": { + "field": "receiver_program", + "type": "bytes32", + "ciphertext": "", + "from": "htlc" + } + }, + "inputs": [ + { + "name": "htlc", + "contract": { + "definition": { + "contractName": "CovenantClaim", + "constructorInputs": [ + { + "name": "preimageHash", + "type": "bytes20" + }, + { + "name": "receiverProgram", + "type": "bytes32" + } + ], + "structs": [], + "functions": [ + { + "name": "claim", + "arkade": { + "inputs": [], + "asm": [ + "OP_PUSHCURRENTINPUTINDEX", + "OP_DUP", + "OP_INSPECTOUTPUTSCRIPTPUBKEY", + "1", + "OP_EQUALVERIFY", + "", + "OP_EQUALVERIFY", + "OP_INSPECTOUTPUTVALUE", + "OP_PUSHCURRENTINPUTINDEX", + "OP_INSPECTINPUTVALUE", + "OP_GREATERTHANOREQUAL" + ] + }, + "leaves": [ + { + "name": "claim", + "asm": [ + "OP_SIZE", + "32", + "OP_EQUALVERIFY", + "OP_HASH160", + "", + "OP_EQUAL", + "OP_VERIFY", + "", + "OP_CHECKSIGVERIFY", + "", + "OP_CHECKSIG" + ], + "witness": [ + { + "name": "preimage", + "type": "bytes32", + "encoding": "raw" + }, + { + "name": "serverSig", + "type": "signature", + "encoding": "schnorr-64", + "injected": true + }, + { + "name": "emulatorSig", + "type": "signature", + "encoding": "schnorr-64", + "injected": true + } + ] + } + ] + } + ] + }, + "arguments": { + "preimageHash": "$( OP_HASH160)", + "receiverProgram": "" + }, + "taptree": "" + }, + "spend": { + "function": "claim", + "leaf": "claim", + "witness": { + "preimage": "" + } + } + } + ], + "outputs": [ + { + "name": "payout", + "index": 0, + "value": { + "from": "htlc" + }, + "locking": "5120$( OP_EQUALVERIFY 0 OP_GREATERTHAN OP_VERIFY 2100000000000000 OP_LESSTHANOREQUAL OP_VERIFY OP_PUSHCURRENTINPUTINDEX OP_INSPECTINPUTVALUE OP_GREATERTHANOREQUAL OP_VERIFY )", + "assets": [ + { + "from": "htlc" + } + ] + } + ], + "packets": { + "output_index": 1, + "unmatched": "drop" + }, + "trigger": { + "packet": 4, + "decode": "tlv", + "min_value": 1, + "require": { + "tlv:3": "", + "tlv:127": "0x64656c6567617465652d736563726574732f7631" + }, + "bind": { + "ciphertext": "tlv:1", + "taptree": "output:taptree", + "receiver_program": { + "from": "tlv:2", + "asm": [ + "OP_PUSHCURRENTINPUTINDEX", + "OP_DUP", + "OP_INSPECTOUTPUTSCRIPTPUBKEY", + "1", + "OP_EQUALVERIFY", + "", + "OP_EQUALVERIFY", + "OP_INSPECTOUTPUTVALUE", + "OP_PUSHCURRENTINPUTINDEX", + "OP_INSPECTINPUTVALUE", + "OP_GREATERTHANOREQUAL" + ] + } + }, + "expires_after_seconds": 86400 + } +} diff --git a/test/e2e/helpers_test.go b/test/e2e/helpers_test.go index 20d3232..79b9670 100644 --- a/test/e2e/helpers_test.go +++ b/test/e2e/helpers_test.go @@ -28,7 +28,6 @@ import ( arklib "github.com/arkade-os/arkd/pkg/ark-lib" "github.com/arkade-os/arkd/pkg/ark-lib/txutils" "github.com/arkade-os/arkd/pkg/client-lib/explorer" - "github.com/arkade-os/arkd/pkg/client-lib/indexer" offchaintx "github.com/arkade-os/arkd/pkg/client-lib/offchain-tx" clientwallet "github.com/arkade-os/arkd/pkg/client-wallet" walletidentity "github.com/arkade-os/arkd/pkg/client-wallet/identity" @@ -39,6 +38,7 @@ import ( "github.com/arkade-os/delegatee/internal/config" "github.com/arkade-os/delegatee/internal/core/application" delegateeexplorer "github.com/arkade-os/delegatee/internal/infrastructure/explorer" + "github.com/arkade-os/delegatee/internal/infrastructure/indexer" grpcservice "github.com/arkade-os/delegatee/internal/interface/grpc" jsontemplate "github.com/arkade-os/delegatee/pkg/template" "github.com/arkade-os/delegatee/pkg/template/ecies" @@ -140,7 +140,11 @@ func delegationsOf(t *testing.T, d delegatee, templateID string) ([]*delegateev1 // faucet pays sats onchain and mines a block. func faucet(t *testing.T, address string, sats int64) { t.Helper() - out, err := exec.Command("nigiri", "faucet", address, btc(sats)).CombinedOutput() + command := exec.Command("nigiri", "faucet", address, btc(sats)) + if cli := os.Getenv("DELEGATEE_TEST_REGTEST_CLI"); cli != "" { + command = exec.Command("node", cli, "faucet", address, btc(sats), "--confirm") + } + out, err := command.CombinedOutput() require.NoError(t, err, string(out)) } @@ -814,17 +818,22 @@ func fundedWallet(t *testing.T) testWallet { t.Cleanup(w.Stop) // Stop panics on a wallet that never initialised log.SetLevel(log.InfoLevel) // the sdk lowers the global level wallet := testWallet{w} - _, _, boarding, err := wallet.Receive(ctx) - require.NoError(t, err) - faucet(t, boarding.Address, 100_000) - - require.Eventually(t, func() bool { - balance, err := wallet.Balance(ctx) - return err == nil && onchainTotal(balance) > 0 - }, 30*time.Second, 500*time.Millisecond, "boarding utxo not detected") - - _, err = wallet.Settle(ctx) - require.NoError(t, err) + if container := os.Getenv("DELEGATEE_TEST_NOTE_CONTAINER"); container != "" { + note, err := exec.Command("docker", "exec", container, "arkd", "note", "--amount", "100000").CombinedOutput() + require.NoError(t, err, string(note)) + _, err = wallet.RedeemNotes(ctx, []string{strings.TrimSpace(string(note))}) + require.NoError(t, err) + } else { + _, _, boarding, err := wallet.Receive(ctx) + require.NoError(t, err) + faucet(t, boarding.Address, 100_000) + require.Eventually(t, func() bool { + balance, err := wallet.Balance(ctx) + return err == nil && onchainTotal(balance) > 0 + }, 30*time.Second, 500*time.Millisecond, "boarding utxo not detected") + _, err = wallet.Settle(ctx) + require.NoError(t, err) + } require.Eventually(t, func() bool { balance, err := wallet.Balance(ctx) return err == nil && balance.OffchainBalance.Total > 0 @@ -850,10 +859,20 @@ func freePort(t *testing.T) uint32 { return uint32(port) } +var ( + arkURL = testEndpoint("DELEGATEE_TEST_ARK_URL", "localhost:7070") + emulatorURL = testEndpoint("DELEGATEE_TEST_EMULATOR_URL", "localhost:7073") + explorerURL = testEndpoint("DELEGATEE_TEST_EXPLORER_URL", "http://localhost:3000") +) + +func testEndpoint(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +} + const ( - arkURL = "localhost:7070" - emulatorURL = "localhost:7073" - explorerURL = "http://localhost:3000" delegateAmount = uint64(10_000) assetAmount = uint64(1_000) password = "password" diff --git a/test/e2e/mutinynet_test.go b/test/e2e/mutinynet_test.go index a58a4d6..21c6a39 100644 --- a/test/e2e/mutinynet_test.go +++ b/test/e2e/mutinynet_test.go @@ -18,8 +18,8 @@ import ( arklib "github.com/arkade-os/arkd/pkg/ark-lib" clientlib "github.com/arkade-os/arkd/pkg/client-lib" arkclient "github.com/arkade-os/arkd/pkg/client-lib/client" - "github.com/arkade-os/arkd/pkg/client-lib/indexer" delegateev1 "github.com/arkade-os/delegatee/api-spec/protobuf/gen/delegatee/v1" + "github.com/arkade-os/delegatee/internal/infrastructure/indexer" "github.com/btcsuite/btcd/btcec/v2" "github.com/stretchr/testify/require" "google.golang.org/protobuf/encoding/protojson" diff --git a/test/e2e/offline_receive_test.go b/test/e2e/offline_receive_test.go new file mode 100644 index 0000000..0b5bba9 --- /dev/null +++ b/test/e2e/offline_receive_test.go @@ -0,0 +1,74 @@ +package e2e + +import ( + "bytes" + "encoding/binary" + "encoding/hex" + "encoding/json" + "strings" + "testing" + + "github.com/arkade-os/arkd/pkg/ark-lib/txutils" + clientlib "github.com/arkade-os/arkd/pkg/client-lib" + offchaintx "github.com/arkade-os/arkd/pkg/client-lib/offchain-tx" + "github.com/arkade-os/delegatee/pkg/template/ecies" + "github.com/arkade-os/delegatee/pkg/template/packets" + "github.com/btcsuite/btcd/psbt/v2" + "github.com/stretchr/testify/require" +) + +func TestOfflineReceivePacketClaim(t *testing.T) { + d := startDelegatee(t) + templateID := fixtureID(t, "offline_receive.json") + streamActive(t, d, templateID) + swap := newSwap(t, d) + addr, pkScript, tree := swapOutput(t, d, swap) + key := d.cfg.EncryptionKeys[0].PubKey() + payload, err := json.Marshal(map[string]any{ + "format": "delegatee-secrets/v1", "template_id": templateID, + "values": map[string]string{"preimage": hex.EncodeToString(swap.preimage), "minimum_value": scriptNum(10_000), "receiver_program": swap.variables["receiver_program"]}, + }) + require.NoError(t, err) + ciphertext, err := ecies.Encrypt(key, payload) + require.NoError(t, err) + program, err := hex.DecodeString(swap.variables["receiver_program"]) + require.NoError(t, err) + marker := []byte("delegatee-secrets/v1") + body := binary.BigEndian.AppendUint16([]byte{127}, uint16(len(marker))) + body = append(body, marker...) + for _, record := range []struct { + tag byte + value []byte + }{{1, ciphertext}, {3, key.SerializeCompressed()}, {2, enforcePayTo(t, program)}} { + body = append(body, record.tag) + body = binary.BigEndian.AppendUint16(body, uint16(len(record.value))) + body = append(body, record.value...) + } + alice := fundedWallet(t) + funding, err := alice.SendOffChain(t.Context(), []clientlib.Receiver{{To: addr, Amount: 10_000}}, + offchaintx.WithExtraPacket(packets.Raw(4, body)), + offchaintx.WithTxOutsTaprootTree(map[string][]byte{hex.EncodeToString(pkScript): tree})) + require.NoError(t, err) + paid := waitForVtxo(t, d, swap.receiverScript, funding) + require.EqualValues(t, 10_000, paid.Amount) + claim, err := d.indexer.GetVirtualTxs(t.Context(), []string{paid.Txid}) + require.NoError(t, err) + require.Len(t, claim.Txs, 1) + ptx, err := psbt.NewFromRawBytes(strings.NewReader(claim.Txs[0]), true) + require.NoError(t, err) + found := false + for i := range ptx.Inputs { + witness, err := txutils.GetArkPsbtConditionWitness(ptx, i) + require.NoError(t, err) + for _, item := range witness { + found = found || bytes.Equal(item, swap.preimage) + } + } + require.True(t, found, "claim condition witness contains the exact preimage") + require.Equal(t, swap.receiverScript, ptx.UnsignedTx.TxOut[0].PkScript) + rows, err := delegationsOf(t, d, templateID) + require.NoError(t, err) + require.Len(t, rows, 1, "funding packet spawned one durable child without client registration") + require.NotZero(t, rows[0].ParentId) + t.Logf("funding=%s claim=%s child=%d receiver=%x", funding, paid.Txid, rows[0].Id, swap.receiverScript) +} diff --git a/test/e2e/restart_test.go b/test/e2e/restart_test.go index a2d63e9..7ec2e72 100644 --- a/test/e2e/restart_test.go +++ b/test/e2e/restart_test.go @@ -13,13 +13,13 @@ import ( arklib "github.com/arkade-os/arkd/pkg/ark-lib" clientlib "github.com/arkade-os/arkd/pkg/client-lib" "github.com/arkade-os/arkd/pkg/client-lib/client" - "github.com/arkade-os/arkd/pkg/client-lib/indexer" delegateev1 "github.com/arkade-os/delegatee/api-spec/protobuf/gen/delegatee/v1" "github.com/arkade-os/delegatee/internal/config" "github.com/arkade-os/delegatee/internal/core/application" "github.com/arkade-os/delegatee/internal/core/ports" "github.com/arkade-os/delegatee/internal/infrastructure/db/postgres" "github.com/arkade-os/delegatee/internal/infrastructure/explorer" + "github.com/arkade-os/delegatee/internal/infrastructure/indexer" emulatorclient "github.com/arkade-os/emulator/pkg/client" "github.com/btcsuite/btcd/btcec/v2/schnorr" "github.com/stretchr/testify/require"