From 7768848b829b36525da8d2b387b4afb867d757e0 Mon Sep 17 00:00:00 2001 From: Rein Krul Date: Fri, 4 Sep 2026 12:09:15 +0200 Subject: [PATCH 1/6] feat(vcr): retry + DLQ for the OpenID4VCI credential-offer push Adds a persistent, retrying queue for OpenID4VCI credential offers that fail on the initial synchronous attempt, and gates the gRPC/DAG network fallback behind it instead of firing on the very first failure. Offer queue (vcr/issuer/offer_queue.go): - Bounded by a fixed 24h wall-clock retry window (tracked from the first failed attempt, surviving restarts via persisted state), not an attempt count - delivery here depends on a remote party's node being reachable, which a local retry cadence can't influence. - Increasing retry interval with jitter, same shape as the existing private-payload-fetch notifier (network/dag/notifier.go). - A pre-first-retry delay: the offer was already attempted once synchronously right before being queued, so the first retry waits rather than immediately re-attempting a very likely still-failing operation (retry-go's own delay only applies *between* attempts). - Once the window is exhausted, the offer is marked dead-lettered (persisted, not removed) and a give-up callback fires - the deferred equivalent of today's immediate fallback. Issue() changes (vcr/issuer/issuer.go): - A genuine delivery failure (OfferCredential itself failing) is now queued instead of immediately falling back to the network: publishing to the DAG is irreversible and network-wide replicated with no dedup, so it shouldn't pay that cost for a failure a retry might resolve. - "Unsupported" cases (no wallet/issuer identifier configured) are unchanged: immediate synchronous fallback, since retrying won't help. - issuer.Issue() no longer guarantees delivery-or-error by the time it returns for the queued case: a 200 means the credential was created and the synchronous attempt was made, not that it was delivered. Documented in the tracking issue's Impact Assessment. Also required: Issuer gained Start()/Shutdown() so the queue's persisted, not-yet-finished offers resume across node restarts, and its in-flight retries stop cleanly on shutdown instead of leaking goroutines. New e2e test (e2e-tests/openid4vci/offer-retry) exercises the real behavior end-to-end: issuing while the receiver is down returns immediately (queued), and the credential is delivered automatically once the receiver comes back - verified locally against a build of this branch, not just via unit tests. The 24h-exhaustion/give-up path is covered by unit tests only; e2e-testing an actual 24h wait isn't practical in CI. Related: #4469 Assisted by AI --- .../openid4vci/offer-retry/docker-compose.yml | 56 ++++ .../openid4vci/offer-retry/node-A/nuts.yaml | 41 +++ .../openid4vci/offer-retry/node-B/nuts.yaml | 41 +++ e2e-tests/openid4vci/offer-retry/run-test.sh | 102 +++++++ e2e-tests/openid4vci/run-tests.sh | 7 + vcr/issuer/interface.go | 6 + vcr/issuer/issuer.go | 75 ++++- vcr/issuer/issuer_test.go | 126 +++++++- vcr/issuer/mock.go | 28 ++ vcr/issuer/offer_queue.go | 249 ++++++++++++++++ vcr/issuer/offer_queue_test.go | 275 ++++++++++++++++++ vcr/vcr.go | 19 +- 12 files changed, 1020 insertions(+), 5 deletions(-) create mode 100644 e2e-tests/openid4vci/offer-retry/docker-compose.yml create mode 100644 e2e-tests/openid4vci/offer-retry/node-A/nuts.yaml create mode 100644 e2e-tests/openid4vci/offer-retry/node-B/nuts.yaml create mode 100755 e2e-tests/openid4vci/offer-retry/run-test.sh create mode 100644 vcr/issuer/offer_queue.go create mode 100644 vcr/issuer/offer_queue_test.go diff --git a/e2e-tests/openid4vci/offer-retry/docker-compose.yml b/e2e-tests/openid4vci/offer-retry/docker-compose.yml new file mode 100644 index 0000000000..a7a8f887fd --- /dev/null +++ b/e2e-tests/openid4vci/offer-retry/docker-compose.yml @@ -0,0 +1,56 @@ +services: + nodeA-backend: + user: "$USER:$USER" + image: "${IMAGE_NODE_A:-nutsfoundation/nuts-node:master}" + environment: + NUTS_CONFIGFILE: /opt/nuts/nuts.yaml + NUTS_NETWORK_NODEDID: "${NODEA_DID}" + ports: + - "18081:8081" + volumes: + - "./node-A/data:/opt/nuts/data" + - "./node-A/nuts.yaml:/opt/nuts/nuts.yaml:ro" + - "../../tls-certs/nodeA-certificate.pem:/opt/nuts/certificate-and-key.pem:ro" + - "../../tls-certs/truststore.pem:/opt/nuts/truststore.pem:ro" + healthcheck: + interval: 1s # Make test run quicker by checking health status more often + nodeA: + image: nginx:1.25.1 + expose: + - 5555 + volumes: + - "../../shared_config/nodeA-grpc-nginx.conf:/etc/nginx/conf.d/nuts-grpc.conf:ro" + - "../../shared_config/nodeA-http-nginx.conf:/etc/nginx/conf.d/nuts-http.conf:ro" + - "../../tls-certs/nodeA-certificate.pem:/etc/nginx/ssl/server.pem:ro" + - "../../tls-certs/nodeA-certificate.pem:/etc/nginx/ssl/key.pem:ro" + - "../../tls-certs/truststore.pem:/etc/nginx/ssl/truststore.pem:ro" + depends_on: + - nodeA-backend + nodeB-backend: + user: "$USER:$USER" + image: "${IMAGE_NODE_B:-nutsfoundation/nuts-node:master}" + environment: + NUTS_CONFIGFILE: /opt/nuts/nuts.yaml + NUTS_NETWORK_NODEDID: "${NODEB_DID}" + NUTS_NETWORK_BOOTSTRAPNODES: ${BOOTSTRAP_NODES} + ports: + - "28081:8081" + volumes: + - "./node-B/data:/opt/nuts/data" + - "./node-B/nuts.yaml:/opt/nuts/nuts.yaml:ro" + - "../../tls-certs/nodeB-certificate.pem:/opt/nuts/certificate-and-key.pem:ro" + - "../../tls-certs/truststore.pem:/opt/nuts/truststore.pem:ro" + healthcheck: + interval: 1s # Make test run quicker by checking health status more often + nodeB: + image: nginx:1.25.1 + expose: + - 5555 + volumes: + - "../../shared_config/nodeB-grpc-nginx.conf:/etc/nginx/conf.d/nuts-grpc.conf:ro" + - "../../shared_config/nodeB-http-nginx.conf:/etc/nginx/conf.d/nuts-http.conf:ro" + - "../../tls-certs/nodeB-certificate.pem:/etc/nginx/ssl/server.pem:ro" + - "../../tls-certs/nodeB-certificate.pem:/etc/nginx/ssl/key.pem:ro" + - "../../tls-certs/truststore.pem:/etc/nginx/ssl/truststore.pem:ro" + depends_on: + - nodeB-backend diff --git a/e2e-tests/openid4vci/offer-retry/node-A/nuts.yaml b/e2e-tests/openid4vci/offer-retry/node-A/nuts.yaml new file mode 100644 index 0000000000..499491ef71 --- /dev/null +++ b/e2e-tests/openid4vci/offer-retry/node-A/nuts.yaml @@ -0,0 +1,41 @@ +url: https://nodeA +verbosity: debug +strictmode: true +internalratelimiter: false +datadir: /opt/nuts/data +http: + internal: + address: :8081 + client: + # Docker auto-assigns the compose network an arbitrary private subnet, so permit all RFC1918 + # ranges for the strict-mode SSRF guard. Narrow-allowlist precision is covered by unit tests. + allowedinternalcidrs: + - 10.0.0.0/8 + - 172.16.0.0/12 + - 192.168.0.0/16 +auth: + contractvalidators: + - dummy + irma: + autoupdateschemas: false +crypto: + storage: fs +goldenhammer: + interval: 1s + enabled: true +vcr: + openid4vci: + enabled: true +tls: + truststorefile: /opt/nuts/truststore.pem + certfile: /opt/nuts/certificate-and-key.pem + certkeyfile: /opt/nuts/certificate-and-key.pem + offload: incoming + certheader: X-SSL-CERT +network: + grpcaddr: :5555 + v2: + gossipinterval: 500 +storage: + sql: + connection: "sqlite:file:/opt/nuts/data/sqlite.db?_pragma=foreign_keys(1)&journal_mode(WAL)" diff --git a/e2e-tests/openid4vci/offer-retry/node-B/nuts.yaml b/e2e-tests/openid4vci/offer-retry/node-B/nuts.yaml new file mode 100644 index 0000000000..1c85067c7b --- /dev/null +++ b/e2e-tests/openid4vci/offer-retry/node-B/nuts.yaml @@ -0,0 +1,41 @@ +url: https://nodeB +verbosity: debug +strictmode: true +internalratelimiter: false +datadir: /opt/nuts/data +http: + internal: + address: :8081 + client: + # Docker auto-assigns the compose network an arbitrary private subnet, so permit all RFC1918 + # ranges for the strict-mode SSRF guard. Narrow-allowlist precision is covered by unit tests. + allowedinternalcidrs: + - 10.0.0.0/8 + - 172.16.0.0/12 + - 192.168.0.0/16 +auth: + contractvalidators: + - dummy + irma: + autoupdateschemas: false +crypto: + storage: fs +goldenhammer: + interval: 1s + enabled: true +vcr: + openid4vci: + enabled: true +tls: + truststorefile: /opt/nuts/truststore.pem + certfile: /opt/nuts/certificate-and-key.pem + certkeyfile: /opt/nuts/certificate-and-key.pem + offload: incoming + certheader: X-SSL-CERT +network: + grpcaddr: :5555 + v2: + gossipinterval: 450 +storage: + sql: + connection: "sqlite:file:/opt/nuts/data/sqlite.db?_pragma=foreign_keys(1)&journal_mode(WAL)" diff --git a/e2e-tests/openid4vci/offer-retry/run-test.sh b/e2e-tests/openid4vci/offer-retry/run-test.sh new file mode 100755 index 0000000000..9c1e5559df --- /dev/null +++ b/e2e-tests/openid4vci/offer-retry/run-test.sh @@ -0,0 +1,102 @@ +#!/usr/bin/env bash +USER=$UID + +set -e + +source ../../util.sh + +echo "------------------------------------" +echo "Cleaning up running Docker containers and volumes, and key material..." +echo "------------------------------------" +# Empty node DIDs to avoid warning in Docker logs +export NODEA_DID= +export NODEB_DID= +export BOOTSTRAP_NODES=nodeA:5555 +docker compose down +docker compose rm -f -v +rm -rf ./node-*/data + +echo "------------------------------------" +echo "Starting Docker containers..." +echo "------------------------------------" +# 'data' dirs will be created with root owner by docker if they do not exist. +# This creates permission issues on CI, since we manually delete the network/connections.db file. +mkdir -p ./node-A/data/network ./node-B/data/network +docker compose up --wait + +echo "------------------------------------" +echo "Creating NodeDIDs, waiting for Golden Hammer to register base URLs..." +echo "------------------------------------" +export NODEA_DID=$(setupNode "http://localhost:18081" "nodeA:5555") +printf "NodeDID for node A: %s\n" "$NODEA_DID" +waitForTXCount "NodeB" "http://localhost:28081/status/diagnostics" 3 10 # 2 for setupNode, 1 for GoldenHammer +export NODEB_DID=$(setupNode "http://localhost:28081" "nodeB:5555") +printf "NodeDID for node B: %s\n" "$NODEB_DID" +waitForTXCount "NodeA" "http://localhost:18081/status/diagnostics" 6 10 # 2 for setupNode, 1 for GoldenHammer + +echo "------------------------------------" +echo "Restarting with NodeDID set..." +echo "------------------------------------" +# Start without bootstrap node, to enforce authenticated, discovered connections +export BOOTSTRAP_NODES= +docker compose exec nodeA-backend rm -f /opt/nuts/data/network/connections.db +docker compose exec nodeB-backend rm -f /opt/nuts/data/network/connections.db +docker compose stop +docker compose up --wait + +echo "------------------------------------" +echo "Stopping node B, to simulate it being (temporarily) unreachable..." +echo "------------------------------------" +docker compose stop nodeB-backend nodeB + +echo "------------------------------------" +echo "Issuing a credential while node B is down..." +echo "------------------------------------" +# The initial synchronous OpenID4VCI push fails (node B is unreachable), but issuing still succeeds +# immediately: the offer is queued for background retry instead of failing the request. +vcNodeA=$(createAuthCredential "http://localhost:18081" "$NODEA_DID" "$NODEB_DID") +printf "VC issued by node A (queued for retry): %s\n" "$vcNodeA" +if [ -z "$vcNodeA" ] || [ "$vcNodeA" == "null" ]; then + echo "FAILED: issuing the credential while node B was down should still succeed immediately (queued for retry)" + exitWithDockerLogs 1 +fi + +echo "------------------------------------" +echo "Bringing node B back up..." +echo "------------------------------------" +docker compose start nodeB-backend nodeB + +echo "------------------------------------" +echo "Waiting for the queued credential to be delivered automatically..." +echo "------------------------------------" +# A longer, dedicated wait: node B needs to fully restart (migrations, etc.) *and* node A's +# background retry needs to fire again, on top of the fixed budget waitForDiagnostic gives elsewhere. +RETRY_TIMEOUT=60 +retry=0 +delivered=false +while [ $retry -lt $RETRY_TIMEOUT ]; do + RESPONSE=$(curl -s "http://localhost:28081/status/diagnostics") + if echo $RESPONSE | grep -q "credential_count: 1"; then + delivered=true + break + fi + printf "." + sleep 1 + retry=$[$retry+1] +done +echo "" +if [ $delivered == false ]; then + echo "FAILED: credential was not delivered to node B within ${RETRY_TIMEOUT}s of it coming back up" + exitWithDockerLogs 1 +fi + +waitForDiagnostic "nodeA-backend" issued_credentials_count 1 + +# Now the credential should be present on both nodeA and nodeB +echo $(readCredential "http://localhost:18081" $vcNodeA) +echo $(readCredential "http://localhost:28081" $vcNodeA) + +echo "------------------------------------" +echo "Stopping Docker containers..." +echo "------------------------------------" +docker compose stop diff --git a/e2e-tests/openid4vci/run-tests.sh b/e2e-tests/openid4vci/run-tests.sh index 4c9cb15c5d..5277a04569 100755 --- a/e2e-tests/openid4vci/run-tests.sh +++ b/e2e-tests/openid4vci/run-tests.sh @@ -16,3 +16,10 @@ echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" pushd network-issuance ./run-test.sh popd + +echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" +echo "!! Running test: Offer Retry !!" +echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" +pushd offer-retry +./run-test.sh +popd diff --git a/vcr/issuer/interface.go b/vcr/issuer/interface.go index 6add1ef3db..631dc32f1a 100644 --- a/vcr/issuer/interface.go +++ b/vcr/issuer/interface.go @@ -53,6 +53,12 @@ type Issuer interface { // GetRevocation returns a revocation for a credential ID. // Returns nil when no revocation is found. GetRevocation(id ssi.URI) (*credential.Revocation, error) + // Start resumes retrying any not-yet-delivered OpenID4VCI credential offers persisted from a + // previous run. Safe to call even if no retry queue is configured. + Start() error + // Shutdown stops any in-flight OpenID4VCI offer retries. Persisted, not-yet-finished offers are + // resumed by the next Start(). + Shutdown() error CredentialSearcher } diff --git a/vcr/issuer/issuer.go b/vcr/issuer/issuer.go index 82f3cfdea7..e0d3b94e5e 100644 --- a/vcr/issuer/issuer.go +++ b/vcr/issuer/issuer.go @@ -26,6 +26,7 @@ import ( "strings" "time" + "github.com/avast/retry-go/v4" "github.com/nuts-foundation/go-stoabs" "github.com/nuts-foundation/nuts-node/vcr/openid4vci" "github.com/nuts-foundation/nuts-node/vcr/revocation" @@ -57,10 +58,13 @@ var TimeFunc = time.Now // since that normally happens through receiving the just-issued credential over the network, // but that doesn't happen when issuing over OpenID4VCI. Thus, it needs to explicitly save it to the VCR store when issuing over OpenID4VCI. // See https://github.com/nuts-foundation/nuts-node/issues/2063 +// offerQueueStore, if non-nil, backs a persistent retry queue for OpenID4VCI credential offers that fail +// on the initial synchronous attempt (see offer_queue.go). If nil, a failed offer falls back to publishing +// over the Nuts network immediately, as if the retry window were already exhausted. func NewIssuer(store Store, vcrStore types.Writer, networkPublisher Publisher, openidHandlerFn func(ctx context.Context, id did.DID) (OpenIDHandler, error), didResolver resolver.DIDResolver, keyStore crypto.KeyStore, jsonldManager jsonld.JSONLD, trustConfig *trust.Config, - statusList *revocation.StatusList2021) Issuer { + statusList *revocation.StatusList2021, offerQueueStore stoabs.KVStore) Issuer { keyResolver := resolver.DIDKeyResolver{Resolver: didResolver} i := &issuer{ store: store, @@ -78,6 +82,9 @@ func NewIssuer(store Store, vcrStore types.Writer, networkPublisher Publisher, } statusList.Sign = i.buildJSONLDCredential statusList.ResolveKey = i.keyResolver.ResolveKey + if offerQueueStore != nil { + i.offerQueue = newOfferQueue(offerQueueStore, i.retryOfferAttempt, i.giveUpOffer) + } return i } @@ -92,6 +99,24 @@ type issuer struct { vcrStore types.Writer walletResolver openid4vci.IdentifierResolver statusList revocation.StatusList2021Issuer + offerQueue *offerQueue +} + +// Start resumes retrying any not-yet-delivered OpenID4VCI credential offers persisted from a previous run. +func (i issuer) Start() error { + if i.offerQueue == nil { + return nil + } + return i.offerQueue.Run() +} + +// Shutdown stops any in-flight OpenID4VCI offer retries. Persisted, not-yet-finished offers are resumed by +// the next Start(). +func (i issuer) Shutdown() error { + if i.offerQueue == nil { + return nil + } + return i.offerQueue.Close() } func (i issuer) GetRevocation(credentialID ssi.URI) (*credential.Revocation, error) { @@ -179,11 +204,21 @@ func (i issuer) Issue(ctx context.Context, template vc.VerifiableCredential, opt if i.openidHandlerFn != nil && !options.Public { success, err := i.issueUsingOpenID4VCI(ctx, *createdVC) if err != nil { - // An error occurred, but it's not because the wallet/issuer doesn't support OpenID4VCI. + // A genuine delivery failure (not "unsupported") - retry in the background instead of + // falling back immediately: publishing to the Nuts network is irreversible and + // network-wide replicated, so it shouldn't pay that cost for a failure that a retry + // might resolve. If no retry queue is configured (offerQueueStore was nil), fall back + // immediately instead, same as before. log.Logger(). WithField(core.LogFieldCredentialID, createdVC.ID.String()). WithError(err). - Warnf("Couldn't publish credential over OpenID4VCI, fallback to publish over Nuts network") + Warnf("Couldn't publish credential over OpenID4VCI, will retry in the background") + if i.offerQueue != nil { + if err := i.offerQueue.Schedule(*createdVC); err != nil { + return nil, fmt.Errorf("unable to queue credential for OpenID4VCI retry: %w", err) + } + return createdVC, nil + } } else if success { log.Logger(). WithField(core.LogFieldCredentialID, createdVC.ID.String()). @@ -229,6 +264,40 @@ func (i issuer) issueUsingOpenID4VCI(ctx context.Context, credential vc.Verifiab return true, i.vcrStore.StoreCredential(credential, nil) } +// retryOfferAttempt adapts issueUsingOpenID4VCI to offerAttemptFn for use by the offer queue: a single +// error return (nil = delivered), and a credential that has since become unsupported (e.g. OpenID4VCI got +// disabled between retries) is treated as unrecoverable rather than retried further. +func (i issuer) retryOfferAttempt(ctx context.Context, credential vc.VerifiableCredential) error { + success, err := i.issueUsingOpenID4VCI(ctx, credential) + if err != nil { + return err + } + if !success { + return retry.Unrecoverable(errOfferNoLongerSupported) + } + return nil +} + +// giveUpOffer is called once an offer's retry window has been exhausted: it's the deferred equivalent of +// the immediate fallback Issue() performs when no retry queue is configured. +func (i issuer) giveUpOffer(ctx context.Context, credential vc.VerifiableCredential) { + log.Logger(). + WithField(core.LogFieldCredentialID, credential.ID.String()). + Warn("Giving up on delivering credential over OpenID4VCI, falling back to publish over the Nuts network") + if i.networkPublisher == nil { + log.Logger(). + WithField(core.LogFieldCredentialID, credential.ID.String()). + Error("No Nuts network publisher configured either; credential delivery has permanently failed") + return + } + if err := i.networkPublisher.PublishCredential(ctx, credential, false); err != nil { + log.Logger(). + WithField(core.LogFieldCredentialID, credential.ID.String()). + WithError(err). + Error("Fallback publish over Nuts network failed after giving up on OpenID4VCI") + } +} + func (i issuer) buildAndSignVC(ctx context.Context, template vc.VerifiableCredential, options CredentialOptions) (*vc.VerifiableCredential, error) { issuerDID, err := did.ParseDID(template.Issuer.String()) if err != nil { diff --git a/vcr/issuer/issuer_test.go b/vcr/issuer/issuer_test.go index 36a038fea9..b7a9badebd 100644 --- a/vcr/issuer/issuer_test.go +++ b/vcr/issuer/issuer_test.go @@ -30,6 +30,7 @@ import ( "gorm.io/gorm" + "github.com/avast/retry-go/v4" "github.com/google/uuid" ssi "github.com/nuts-foundation/go-did" "github.com/nuts-foundation/go-did/did" @@ -370,6 +371,50 @@ func Test_issuer_Issue(t *testing.T) { require.NoError(t, err) assert.NotNil(t, result) }) + t.Run("ok - publish over OpenID4VCI fails - queued for retry instead of immediate fallback", func(t *testing.T) { + ctrl := gomock.NewController(t) + // No PublishCredential call expected: a genuine delivery failure is queued, not fallen back + // to immediately, when a retry queue is configured. + publisher := NewMockPublisher(ctrl) + walletResolver := openid4vci.NewMockIdentifierResolver(ctrl) + walletResolver.EXPECT().Resolve(gomock.Any()).Return(walletIdentifier, nil) + openidHandler := NewMockOpenIDHandler(ctrl) + openidHandler.EXPECT().OfferCredential(gomock.Any(), gomock.Any(), walletIdentifier).Return(errors.New("failed")) + keyResolverMock := resolver.NewMockKeyResolver(ctrl) + keyResolverMock.EXPECT().ResolveKey(issuerDID, nil, resolver.AssertionMethod).Return(issuerKeyID, issuerKey, nil) + store := NewMockStore(ctrl) + store.EXPECT().StoreCredential(gomock.Any()) + sut := issuer{ + keyResolver: keyResolverMock, + store: store, + jsonldManager: jsonldManager, + trustConfig: trust.NewConfig(path.Join(io.TestDirectory(t), "trust.config")), + keyStore: nutsCryptoInstance, + openidHandlerFn: func(_ context.Context, id did.DID) (OpenIDHandler, error) { + if id.Equals(issuerDID) { + return openidHandler, nil + } + return nil, nil + }, + walletResolver: walletResolver, + networkPublisher: publisher, + } + sut.offerQueue = newOfferQueue(testOfferQueueStore(t), sut.retryOfferAttempt, sut.giveUpOffer) + t.Cleanup(func() { _ = sut.offerQueue.Close() }) + + result, err := sut.Issue(ctx, template, CredentialOptions{ + Publish: true, + Public: false, + }) + + require.NoError(t, err) + assert.NotNil(t, result) + + jobs, err := sut.offerQueue.all() + require.NoError(t, err) + require.Len(t, jobs, 1) + require.Equal(t, result.ID.String(), jobs[0].Credential.ID.String()) + }) t.Run("ok - OpenID4VCI not enabled - fallback to network", func(t *testing.T) { ctrl := gomock.NewController(t) publisher := NewMockPublisher(ctrl) @@ -539,10 +584,89 @@ func Test_issuer_Issue(t *testing.T) { } func TestNewIssuer(t *testing.T) { - createdIssuer := NewIssuer(nil, nil, nil, nil, nil, nil, nil, nil, &revocation.StatusList2021{}) + createdIssuer := NewIssuer(nil, nil, nil, nil, nil, nil, nil, nil, &revocation.StatusList2021{}, nil) assert.IsType(t, &issuer{}, createdIssuer) } +func Test_issuer_retryOfferAttempt(t *testing.T) { + const walletIdentifier = "http://example.com/wallet" + credential := testOfferQueueCredential(t, "did:nuts:issuer#retry-1") + + t.Run("ok - delivered", func(t *testing.T) { + ctrl := gomock.NewController(t) + walletResolver := openid4vci.NewMockIdentifierResolver(ctrl) + walletResolver.EXPECT().Resolve(gomock.Any()).Return(walletIdentifier, nil) + openidHandler := NewMockOpenIDHandler(ctrl) + openidHandler.EXPECT().OfferCredential(gomock.Any(), gomock.Any(), walletIdentifier).Return(nil) + vcrStore := vcr.NewMockWriter(ctrl) + vcrStore.EXPECT().StoreCredential(gomock.Any(), gomock.Any()) + sut := issuer{ + walletResolver: walletResolver, + openidHandlerFn: func(_ context.Context, _ did.DID) (OpenIDHandler, error) { + return openidHandler, nil + }, + vcrStore: vcrStore, + } + + err := sut.retryOfferAttempt(context.Background(), credential) + + require.NoError(t, err) + }) + + t.Run("error - genuine delivery failure is returned as-is (retryable)", func(t *testing.T) { + ctrl := gomock.NewController(t) + walletResolver := openid4vci.NewMockIdentifierResolver(ctrl) + walletResolver.EXPECT().Resolve(gomock.Any()).Return(walletIdentifier, nil) + openidHandler := NewMockOpenIDHandler(ctrl) + openidHandler.EXPECT().OfferCredential(gomock.Any(), gomock.Any(), walletIdentifier).Return(errors.New("still failing")) + sut := issuer{ + walletResolver: walletResolver, + openidHandlerFn: func(_ context.Context, _ did.DID) (OpenIDHandler, error) { + return openidHandler, nil + }, + } + + err := sut.retryOfferAttempt(context.Background(), credential) + + require.Error(t, err) + assert.True(t, retry.IsRecoverable(err), "a plain delivery error must remain retryable") + }) + + t.Run("ok - became unsupported between retries is unrecoverable", func(t *testing.T) { + ctrl := gomock.NewController(t) + walletResolver := openid4vci.NewMockIdentifierResolver(ctrl) + walletResolver.EXPECT().Resolve(gomock.Any()).Return("", nil) // wallet no longer configured + sut := issuer{ + walletResolver: walletResolver, + } + + err := sut.retryOfferAttempt(context.Background(), credential) + + require.Error(t, err) + assert.ErrorIs(t, err, errOfferNoLongerSupported) + assert.False(t, retry.IsRecoverable(err), "should be unrecoverable, so the queue gives up instead of retrying further") + }) +} + +func Test_issuer_giveUpOffer(t *testing.T) { + credential := testOfferQueueCredential(t, "did:nuts:issuer#retry-2") + + t.Run("falls back to the Nuts network", func(t *testing.T) { + ctrl := gomock.NewController(t) + publisher := NewMockPublisher(ctrl) + publisher.EXPECT().PublishCredential(gomock.Any(), gomock.Any(), false).Return(nil) + sut := issuer{networkPublisher: publisher} + + sut.giveUpOffer(context.Background(), credential) + }) + + t.Run("no network publisher configured - logs and does not panic", func(t *testing.T) { + sut := issuer{networkPublisher: nil} + + sut.giveUpOffer(context.Background(), credential) + }) +} + func Test_issuer_buildRevocation(t *testing.T) { jsonldManager := jsonld.NewTestJSONLDManager(t) ctx := audit.TestContext() diff --git a/vcr/issuer/mock.go b/vcr/issuer/mock.go index 888da1919a..0fbdfc3a80 100644 --- a/vcr/issuer/mock.go +++ b/vcr/issuer/mock.go @@ -157,6 +157,34 @@ func (mr *MockIssuerMockRecorder) SearchCredential(credentialType, issuer, subje return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SearchCredential", reflect.TypeOf((*MockIssuer)(nil).SearchCredential), credentialType, issuer, subject) } +// Shutdown mocks base method. +func (m *MockIssuer) Shutdown() error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Shutdown") + ret0, _ := ret[0].(error) + return ret0 +} + +// Shutdown indicates an expected call of Shutdown. +func (mr *MockIssuerMockRecorder) Shutdown() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Shutdown", reflect.TypeOf((*MockIssuer)(nil).Shutdown)) +} + +// Start mocks base method. +func (m *MockIssuer) Start() error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Start") + ret0, _ := ret[0].(error) + return ret0 +} + +// Start indicates an expected call of Start. +func (mr *MockIssuerMockRecorder) Start() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockIssuer)(nil).Start)) +} + // StatusList mocks base method. func (m *MockIssuer) StatusList(ctx context.Context, issuer did.DID, page int) (*vc.VerifiableCredential, error) { m.ctrl.T.Helper() diff --git a/vcr/issuer/offer_queue.go b/vcr/issuer/offer_queue.go new file mode 100644 index 0000000000..db9632c8fd --- /dev/null +++ b/vcr/issuer/offer_queue.go @@ -0,0 +1,249 @@ +/* + * Copyright (C) 2026 Nuts community + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + * + */ + +package issuer + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/avast/retry-go/v4" + "github.com/nuts-foundation/go-did/vc" + "github.com/nuts-foundation/go-stoabs" + "github.com/nuts-foundation/nuts-node/core" + "github.com/nuts-foundation/nuts-node/vcr/log" +) + +const offerQueueShelfName = "openid4vci_offer_queue" + +// offerRetryWindow is the maximum total wall-clock time an offer is retried for, counted from the first +// failed attempt (across restarts, since it's read from persisted state), before the offer is considered +// dead-lettered and the give-up callback is invoked. Modeled on the private-payload-fetch notifier +// (network/dag/notifier.go), which bounds retries the same way, but by attempt count within a delay cap +// rather than a fixed wall-clock window; a window fits this use case better since delivery here depends on +// a remote party's node being reachable/fixed, not on local retry cadence. +var offerRetryWindow = 24 * time.Hour + +// offerRetryInitialDelay is the delay before the first retry; it then increases (with jitter) after every +// subsequent failure, capped at offerRetryMaxDelay. Matches the starting delay already used by +// network/dag/notifier.go's default. +var offerRetryInitialDelay = time.Second + +// offerRetryMaxDelay caps the delay between individual retry attempts, so a job still checks in reasonably +// often across the full offerRetryWindow instead of the delay growing unbounded. +var offerRetryMaxDelay = time.Hour + +// offerAttemptFn attempts to deliver a single credential offer. Returning nil means delivery succeeded. +type offerAttemptFn func(ctx context.Context, credential vc.VerifiableCredential) error + +// offerGiveUpFn is called exactly once, when an offer's retry window has been exhausted without success. +type offerGiveUpFn func(ctx context.Context, credential vc.VerifiableCredential) + +// errOfferNoLongerSupported signals that OpenID4VCI is no longer usable for this offer (e.g. the wallet or +// issuer stopped supporting it between retries) and that retrying further won't help. +var errOfferNoLongerSupported = errors.New("wallet or issuer no longer supports OpenID4VCI") + +// offerJob is the persisted state of a single retrying credential offer. +type offerJob struct { + Credential vc.VerifiableCredential `json:"credential"` + FirstAttempt time.Time `json:"firstAttempt"` + Retries int `json:"retries"` + Latest *time.Time `json:"latest,omitempty"` + Error string `json:"error,omitempty"` + // GivenUp indicates the retry window was exhausted; the offer is dead-lettered. + GivenUp bool `json:"givenUp,omitempty"` +} + +// offerQueue is a persistent, retrying queue for OpenID4VCI credential offers that failed on the initial +// synchronous attempt. Modeled on network/dag's private-payload-fetch notifier: durable per-job state, +// exponential backoff via retry-go, but bounded by a fixed total retry window rather than an attempt count. +type offerQueue struct { + db stoabs.KVStore + attempt offerAttemptFn + giveUp offerGiveUpFn + ctx context.Context + cancel context.CancelFunc +} + +// newOfferQueue creates an offerQueue backed by db. attempt is called for every (re)try; giveUp is called +// once when an offer's retry window is exhausted. +func newOfferQueue(db stoabs.KVStore, attempt offerAttemptFn, giveUp offerGiveUpFn) *offerQueue { + ctx, cancel := context.WithCancel(context.Background()) + return &offerQueue{ + db: db, + attempt: attempt, + giveUp: giveUp, + ctx: ctx, + cancel: cancel, + } +} + +// Schedule persists the credential and starts retrying its offer in the background. +func (q *offerQueue) Schedule(credential vc.VerifiableCredential) error { + job := offerJob{ + Credential: credential, + FirstAttempt: time.Now(), + } + if err := q.save(job); err != nil { + return err + } + go q.retry(job) + return nil +} + +// Run resumes retrying every persisted offer that hasn't given up yet. Call once at startup. +func (q *offerQueue) Run() error { + jobs, err := q.all() + if err != nil { + return err + } + for _, job := range jobs { + if job.GivenUp { + continue + } + go q.retry(job) + } + return nil +} + +// GetFailedOffers returns offers whose retry window has been exhausted (dead-lettered). +func (q *offerQueue) GetFailedOffers() ([]offerJob, error) { + jobs, err := q.all() + if err != nil { + return nil, err + } + var failed []offerJob + for _, job := range jobs { + if job.GivenUp { + failed = append(failed, job) + } + } + return failed, nil +} + +// Close stops all in-flight retries. Persisted jobs are left untouched; Run() picks them back up on the +// next startup. +func (q *offerQueue) Close() error { + q.cancel() + return nil +} + +func (q *offerQueue) retry(job offerJob) { + deadline := job.FirstAttempt.Add(offerRetryWindow) + ctx, cancel := context.WithDeadline(q.ctx, deadline) + defer cancel() + + // retry.Do calls the given function immediately on its first attempt; delay only applies *between* + // attempts. But the offer was already attempted once synchronously, right before it was scheduled + // (that's why it's here), so wait before this first retry rather than immediately re-attempting a + // very likely still-failing operation. + select { + case <-time.After(offerRetryInitialDelay): + case <-ctx.Done(): + q.settle(job, ctx.Err()) + return + } + + err := retry.Do(func() error { + return q.attempt(ctx, job.Credential) + }, + retry.Context(ctx), + retry.Attempts(0), // unbounded attempts; the context deadline is the real bound + retry.Delay(offerRetryInitialDelay), + retry.MaxDelay(offerRetryMaxDelay), + retry.MaxJitter(offerRetryInitialDelay), + retry.DelayType(retry.CombineDelay(retry.BackOffDelay, retry.RandomDelay)), + retry.LastErrorOnly(true), + retry.OnRetry(func(n uint, retryErr error) { + job.Retries++ + now := time.Now() + job.Latest = &now + job.Error = retryErr.Error() + if saveErr := q.save(job); saveErr != nil { + log.Logger().WithError(saveErr).Warn("Failed to persist OpenID4VCI offer retry state") + } + log.Logger(). + WithError(retryErr). + WithField(core.LogFieldCredentialID, job.Credential.ID.String()). + Debugf("Retrying OpenID4VCI credential offer (attempt %d)", n) + }), + ) + q.settle(job, err) +} + +// settle handles the outcome of a retry() run, whether from retry.Do itself or from the pre-first-attempt +// wait being cancelled before ever calling attempt. +func (q *offerQueue) settle(job offerJob, err error) { + if err == nil { + if finishErr := q.finish(job); finishErr != nil { + log.Logger().WithError(finishErr).Warn("Failed to remove finished OpenID4VCI offer from retry queue") + } + return + } + if errors.Is(q.ctx.Err(), context.Canceled) { + // Queue was closed (e.g. node shutting down), not the job's own deadline. Leave it persisted as-is; + // Run() resumes it on the next startup, still counting from its original FirstAttempt. + return + } + // Either the retry window (24h) was exhausted, or the offer became unsupported (errOfferNoLongerSupported). + job.GivenUp = true + now := time.Now() + job.Latest = &now + job.Error = err.Error() + if saveErr := q.save(job); saveErr != nil { + log.Logger().WithError(saveErr).Warn("Failed to persist OpenID4VCI offer as given up") + } + q.giveUp(q.ctx, job.Credential) +} + +// Persistence operations deliberately use context.Background() rather than q.ctx: q.ctx is cancelled by +// Close() to stop in-flight retries, but reading/writing the persisted queue (e.g. from GetFailedOffers(), +// callable independently of whether the queue is still running) must keep working regardless. + +func (q *offerQueue) save(job offerJob) error { + data, err := json.Marshal(job) + if err != nil { + return err + } + return q.db.WriteShelf(context.Background(), offerQueueShelfName, func(writer stoabs.Writer) error { + return writer.Put(stoabs.BytesKey(job.Credential.ID.String()), data) + }) +} + +func (q *offerQueue) finish(job offerJob) error { + return q.db.WriteShelf(context.Background(), offerQueueShelfName, func(writer stoabs.Writer) error { + return writer.Delete(stoabs.BytesKey(job.Credential.ID.String())) + }) +} + +func (q *offerQueue) all() ([]offerJob, error) { + var jobs []offerJob + err := q.db.ReadShelf(context.Background(), offerQueueShelfName, func(reader stoabs.Reader) error { + return reader.Iterate(func(_ stoabs.Key, v []byte) error { + var job offerJob + if err := json.Unmarshal(v, &job); err != nil { + return err + } + jobs = append(jobs, job) + return nil + }, stoabs.BytesKey{}) + }) + return jobs, err +} diff --git a/vcr/issuer/offer_queue_test.go b/vcr/issuer/offer_queue_test.go new file mode 100644 index 0000000000..8a4735931d --- /dev/null +++ b/vcr/issuer/offer_queue_test.go @@ -0,0 +1,275 @@ +/* + * Copyright (C) 2026 Nuts community + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + * + */ + +package issuer + +import ( + "context" + "errors" + "path" + "sync/atomic" + "testing" + "time" + + "github.com/avast/retry-go/v4" + ssi "github.com/nuts-foundation/go-did" + "github.com/nuts-foundation/go-did/vc" + "github.com/nuts-foundation/go-stoabs" + "github.com/nuts-foundation/go-stoabs/bbolt" + "github.com/stretchr/testify/require" +) + +func testOfferQueueCredential(t *testing.T, id string) vc.VerifiableCredential { + uri := ssi.MustParseURI(id) + return vc.VerifiableCredential{ + ID: &uri, + Issuer: ssi.MustParseURI("did:nuts:issuer"), + CredentialSubject: []map[string]any{{"id": "did:nuts:holder"}}, + } +} + +func testOfferQueueStore(t *testing.T) stoabs.KVStore { + dbPath := path.Join(t.TempDir(), "offer_queue.db") + db, err := bbolt.CreateBBoltStore(dbPath) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close(context.Background()) }) + return db +} + +// withFastRetryTiming overrides the package-level retry timing vars for the duration of the test. +func withFastRetryTiming(t *testing.T, window time.Duration) { + originalWindow, originalInitial, originalMax := offerRetryWindow, offerRetryInitialDelay, offerRetryMaxDelay + offerRetryWindow = window + offerRetryInitialDelay = time.Millisecond + offerRetryMaxDelay = 10 * time.Millisecond + t.Cleanup(func() { + offerRetryWindow, offerRetryInitialDelay, offerRetryMaxDelay = originalWindow, originalInitial, originalMax + }) +} + +func TestOfferQueue_Schedule(t *testing.T) { + t.Run("succeeds on first attempt", func(t *testing.T) { + withFastRetryTiming(t, time.Second) + db := testOfferQueueStore(t) + var attempts atomic.Int32 + done := make(chan struct{}) + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + attempts.Add(1) + close(done) + return nil + }, + func(_ context.Context, _ vc.VerifiableCredential) { t.Fatal("giveUp should not be called") }, + ) + t.Cleanup(func() { _ = q.Close() }) + + require.NoError(t, q.Schedule(testOfferQueueCredential(t, "did:nuts:issuer#1"))) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for attempt") + } + require.Equal(t, int32(1), attempts.Load()) + + // Finished offers are removed from the persisted queue. + require.Eventually(t, func() bool { + jobs, err := q.all() + require.NoError(t, err) + return len(jobs) == 0 + }, time.Second, 10*time.Millisecond) + }) + + t.Run("retries after a failure, then succeeds", func(t *testing.T) { + withFastRetryTiming(t, time.Second) + db := testOfferQueueStore(t) + var attempts atomic.Int32 + done := make(chan struct{}) + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + if attempts.Add(1) < 3 { + return errors.New("transient failure") + } + close(done) + return nil + }, + func(_ context.Context, _ vc.VerifiableCredential) { t.Fatal("giveUp should not be called") }, + ) + t.Cleanup(func() { _ = q.Close() }) + + require.NoError(t, q.Schedule(testOfferQueueCredential(t, "did:nuts:issuer#2"))) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for eventual success") + } + require.GreaterOrEqual(t, attempts.Load(), int32(3)) + }) + + t.Run("gives up once the retry window is exhausted", func(t *testing.T) { + // A generous window relative to the 1ms initial delay set by withFastRetryTiming: needs enough + // margin that at least one attempt reliably completes before the deadline, even under scheduling + // jitter/CPU contention in CI, while still keeping the test itself fast. + withFastRetryTiming(t, 500*time.Millisecond) + db := testOfferQueueStore(t) + var attempts atomic.Int32 + givenUp := make(chan vc.VerifiableCredential, 1) + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + attempts.Add(1) + return errors.New("permanent failure") + }, + func(_ context.Context, credential vc.VerifiableCredential) { + givenUp <- credential + }, + ) + t.Cleanup(func() { _ = q.Close() }) + + credential := testOfferQueueCredential(t, "did:nuts:issuer#3") + require.NoError(t, q.Schedule(credential)) + + select { + case got := <-givenUp: + require.Equal(t, credential.ID.String(), got.ID.String()) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for give-up") + } + require.Greater(t, attempts.Load(), int32(0)) + + // The dead-lettered job stays persisted, marked as given up. + failed, err := q.GetFailedOffers() + require.NoError(t, err) + require.Len(t, failed, 1) + require.True(t, failed[0].GivenUp) + require.Equal(t, credential.ID.String(), failed[0].Credential.ID.String()) + }) + + t.Run("an unrecoverable error stops retrying immediately", func(t *testing.T) { + // The queue itself has no special-cased errors; it's the caller's job to wrap an error with + // retry.Unrecoverable() to signal "don't bother retrying" (this is exactly what + // issuer.retryOfferAttempt does for errOfferNoLongerSupported - see issuer_test.go for that). + withFastRetryTiming(t, time.Second) + db := testOfferQueueStore(t) + var attempts atomic.Int32 + givenUp := make(chan struct{}) + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + attempts.Add(1) + return retry.Unrecoverable(errOfferNoLongerSupported) + }, + func(_ context.Context, _ vc.VerifiableCredential) { close(givenUp) }, + ) + t.Cleanup(func() { _ = q.Close() }) + + require.NoError(t, q.Schedule(testOfferQueueCredential(t, "did:nuts:issuer#4"))) + + select { + case <-givenUp: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for give-up") + } + require.Equal(t, int32(1), attempts.Load()) + }) +} + +func TestOfferQueue_Run(t *testing.T) { + t.Run("resumes a persisted job from a previous run", func(t *testing.T) { + withFastRetryTiming(t, time.Second) + db := testOfferQueueStore(t) + credential := testOfferQueueCredential(t, "did:nuts:issuer#5") + + // Simulate state left behind by a previous process, as if Schedule() had run then the process + // stopped before the job finished. + bootstrapQueue := newOfferQueue(db, nil, nil) + require.NoError(t, bootstrapQueue.save(offerJob{Credential: credential, FirstAttempt: time.Now()})) + + done := make(chan struct{}) + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + close(done) + return nil + }, + func(_ context.Context, _ vc.VerifiableCredential) { t.Fatal("giveUp should not be called") }, + ) + t.Cleanup(func() { _ = q.Close() }) + + require.NoError(t, q.Run()) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for resumed job to be retried") + } + }) + + t.Run("does not resume a job that already gave up", func(t *testing.T) { + withFastRetryTiming(t, time.Second) + db := testOfferQueueStore(t) + credential := testOfferQueueCredential(t, "did:nuts:issuer#6") + + bootstrapQueue := newOfferQueue(db, nil, nil) + require.NoError(t, bootstrapQueue.save(offerJob{Credential: credential, FirstAttempt: time.Now(), GivenUp: true})) + + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + t.Fatal("attempt should not be called for an already given-up job") + return nil + }, + func(_ context.Context, _ vc.VerifiableCredential) { t.Fatal("giveUp should not be called again") }, + ) + t.Cleanup(func() { _ = q.Close() }) + + require.NoError(t, q.Run()) + time.Sleep(50 * time.Millisecond) // give any (unwanted) goroutine a chance to run + }) +} + +func TestOfferQueue_Close(t *testing.T) { + t.Run("stops in-flight retries without marking the job given up", func(t *testing.T) { + withFastRetryTiming(t, time.Second) + db := testOfferQueueStore(t) + attempted := make(chan struct{}, 10) + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + select { + case attempted <- struct{}{}: + default: + } + return errors.New("still failing") + }, + func(_ context.Context, _ vc.VerifiableCredential) { t.Fatal("giveUp should not be called on shutdown") }, + ) + + credential := testOfferQueueCredential(t, "did:nuts:issuer#7") + require.NoError(t, q.Schedule(credential)) + + select { + case <-attempted: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for first attempt") + } + require.NoError(t, q.Close()) + time.Sleep(50 * time.Millisecond) // let the retry goroutine observe cancellation + + jobs, err := q.all() + require.NoError(t, err) + require.Len(t, jobs, 1) + require.False(t, jobs[0].GivenUp) + }) +} diff --git a/vcr/vcr.go b/vcr/vcr.go index 7804c5f1e1..1fce19dd3f 100644 --- a/vcr/vcr.go +++ b/vcr/vcr.go @@ -25,6 +25,7 @@ import ( "errors" "fmt" "github.com/nuts-foundation/go-leia/v4" + "github.com/nuts-foundation/go-stoabs" "github.com/nuts-foundation/nuts-node/http/client" "github.com/nuts-foundation/nuts-node/pki" "github.com/nuts-foundation/nuts-node/vcr/credential" @@ -229,8 +230,16 @@ func (c *vcr) Configure(config core.ServerConfig) error { networkPublisher = issuer.NewNetworkPublisher(c.network, didResolver, c.keyStore) } + var offerQueueStore stoabs.KVStore + if c.config.OpenID4VCI.Enabled { + offerQueueStore, err = c.storageClient.GetProvider(ModuleName).GetKVStore("openid4vci-offer-queue", storage.PersistentStorageClass) + if err != nil { + return err + } + } + status := revocation.NewStatusList2021(c.storageClient.GetSQLDatabase(), client.NewWithCache(config.HTTPClient.Timeout), config.URL) - c.issuer = issuer.NewIssuer(c.issuerStore, c, networkPublisher, openidHandlerFn, didResolver, c.keyStore, c.jsonldManager, c.trustConfig, status) + c.issuer = issuer.NewIssuer(c.issuerStore, c, networkPublisher, openidHandlerFn, didResolver, c.keyStore, c.jsonldManager, c.trustConfig, status, offerQueueStore) c.verifier = verifier.NewVerifier(c.verifierStore, didResolver, c.keyResolver, c.jsonldManager, c.trustConfig, status, c.pkiProvider) if !c.network.Disabled() { @@ -275,6 +284,9 @@ func (c *vcr) createCredentialsStore() error { } func (c *vcr) Start() error { + if err := c.issuer.Start(); err != nil { + return err + } if c.ambassador == nil { // did:nuts / network layer is disabled return nil } @@ -285,6 +297,11 @@ func (c *vcr) Start() error { } func (c *vcr) Shutdown() error { + if err := c.issuer.Shutdown(); err != nil { + log.Logger(). + WithError(err). + Error("Unable to shut down issuer") + } err := c.issuerStore.Close() if err != nil { log.Logger(). From aec04cbf0a52f5c1703b8def3f7cd1f5fa4bba39 Mon Sep 17 00:00:00 2001 From: Rein Krul Date: Fri, 4 Sep 2026 12:31:57 +0200 Subject: [PATCH 2/6] docs(vcr): trim Start/Shutdown godoc, clarify offerQueueStore nilability Start/Shutdown just delegate to offerQueue.Run()/Close(), which already document the resume/persist behavior; restate only that. offerQueueStore's doc previously implied its nilness was an independent operational fallback mode - in fact it's nil exactly when openidHandlerFn is nil (both gated by the same OpenID4VCI.Enabled check in vcr.go), and is only independently nilable to support constructing an issuer without a queue in tests. Assisted by AI --- vcr/issuer/issuer.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/vcr/issuer/issuer.go b/vcr/issuer/issuer.go index e0d3b94e5e..2c231a2e29 100644 --- a/vcr/issuer/issuer.go +++ b/vcr/issuer/issuer.go @@ -58,9 +58,10 @@ var TimeFunc = time.Now // since that normally happens through receiving the just-issued credential over the network, // but that doesn't happen when issuing over OpenID4VCI. Thus, it needs to explicitly save it to the VCR store when issuing over OpenID4VCI. // See https://github.com/nuts-foundation/nuts-node/issues/2063 -// offerQueueStore, if non-nil, backs a persistent retry queue for OpenID4VCI credential offers that fail -// on the initial synchronous attempt (see offer_queue.go). If nil, a failed offer falls back to publishing -// over the Nuts network immediately, as if the retry window were already exhausted. +// offerQueueStore backs the persistent OpenID4VCI offer retry queue (see offer_queue.go). It's nil exactly +// when openidHandlerFn is nil in the current (only) caller, vcr.go, since both are gated by the same +// OpenID4VCI.Enabled check; NewIssuer accepts it as a separate, independently-nilable parameter so tests can +// construct an issuer without a queue. func NewIssuer(store Store, vcrStore types.Writer, networkPublisher Publisher, openidHandlerFn func(ctx context.Context, id did.DID) (OpenIDHandler, error), didResolver resolver.DIDResolver, keyStore crypto.KeyStore, jsonldManager jsonld.JSONLD, trustConfig *trust.Config, @@ -102,7 +103,7 @@ type issuer struct { offerQueue *offerQueue } -// Start resumes retrying any not-yet-delivered OpenID4VCI credential offers persisted from a previous run. +// Start resumes the OpenID4VCI offer retry queue, if configured; see offerQueue.Run(). func (i issuer) Start() error { if i.offerQueue == nil { return nil @@ -110,8 +111,7 @@ func (i issuer) Start() error { return i.offerQueue.Run() } -// Shutdown stops any in-flight OpenID4VCI offer retries. Persisted, not-yet-finished offers are resumed by -// the next Start(). +// Shutdown stops the OpenID4VCI offer retry queue, if configured; see offerQueue.Close(). func (i issuer) Shutdown() error { if i.offerQueue == nil { return nil From c4d849650ac9efcab53938ad0c2c935212d5e81e Mon Sep 17 00:00:00 2001 From: Rein Krul Date: Fri, 4 Sep 2026 12:34:25 +0200 Subject: [PATCH 3/6] test(e2e): assert offer-retry delivery never touches the DAG Previously only checked the credential arrived, which is consistent with either delivery path. Assert node A's transaction_count is unchanged before and after delivery, proving it went over OpenID4VCI (offerQueue never calls the DAG publisher while retrying) rather than the gRPC/DAG fallback. Verified locally: built the branch's image, ran the test end to end (exit 0), transaction_count stayed at 6 across the retry. Assisted by AI --- e2e-tests/openid4vci/offer-retry/run-test.sh | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/e2e-tests/openid4vci/offer-retry/run-test.sh b/e2e-tests/openid4vci/offer-retry/run-test.sh index 9c1e5559df..28d743c5dc 100755 --- a/e2e-tests/openid4vci/offer-retry/run-test.sh +++ b/e2e-tests/openid4vci/offer-retry/run-test.sh @@ -44,6 +44,12 @@ docker compose exec nodeB-backend rm -f /opt/nuts/data/network/connections.db docker compose stop docker compose up --wait +echo "------------------------------------" +echo "Recording node A's transaction count baseline..." +echo "------------------------------------" +txCountBefore=$(readDiagnostic "http://localhost:18081" "transaction_count") +printf "Node A transaction_count before issuance: %s\n" "$txCountBefore" + echo "------------------------------------" echo "Stopping node B, to simulate it being (temporarily) unreachable..." echo "------------------------------------" @@ -92,6 +98,18 @@ fi waitForDiagnostic "nodeA-backend" issued_credentials_count 1 +echo "------------------------------------" +echo "Verifying delivery went over OpenID4VCI, not the Nuts network (DAG) fallback..." +echo "------------------------------------" +# The offer was only ever retried over OpenID4VCI (never published to the DAG, see offer_queue.go), so node +# A's transaction count must be unchanged: a gRPC/DAG-delivered credential would have added a transaction. +txCountAfter=$(readDiagnostic "http://localhost:18081" "transaction_count") +printf "Node A transaction_count after delivery: %s\n" "$txCountAfter" +if [ "$txCountAfter" != "$txCountBefore" ]; then + echo "FAILED: node A's transaction count changed ($txCountBefore -> $txCountAfter); credential appears to have been published to the DAG instead of delivered via OpenID4VCI retry" + exitWithDockerLogs 1 +fi + # Now the credential should be present on both nodeA and nodeB echo $(readCredential "http://localhost:18081" $vcNodeA) echo $(readCredential "http://localhost:28081" $vcNodeA) From b81b112471821c6b5d5a727736ee875e0b0eb0eb Mon Sep 17 00:00:00 2001 From: Rein Krul Date: Fri, 4 Sep 2026 13:57:38 +0200 Subject: [PATCH 4/6] fix(vcr): preserve the real delivery error when an offer is dead-lettered retry.Do() was missing WrapContextErrorWithLastError(true): when the 24h offerRetryWindow expires between attempts (the normal give-up path), it returned a bare context.DeadlineExceeded and discarded the actual last delivery error, so a dead-lettered job's persisted Error just said "context deadline exceeded" - useless for the diagnostics/requeue tooling planned in #4469 item 3. Verified against the pinned retry-go v4.7.0 source: without the option, Do() returns context.Cause(ctx) alone on the ctx.Done() branch of its unbounded-attempts loop. Also fixed the same class of loss in retry()'s pre-wait bail-out (when the deadline hits before retry.Do is even called): it now folds any previously-recorded job.Error into the returned error instead of overwriting it with a bare context error. Strengthened the existing "gives up once the retry window is exhausted" test to assert on Error content, which the prior version never checked - verified it fails without this fix (bare "context deadline exceeded"), passes with it. Assisted by AI --- vcr/issuer/offer_queue.go | 14 +++++++++++++- vcr/issuer/offer_queue_test.go | 6 +++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/vcr/issuer/offer_queue.go b/vcr/issuer/offer_queue.go index db9632c8fd..210e8e1209 100644 --- a/vcr/issuer/offer_queue.go +++ b/vcr/issuer/offer_queue.go @@ -22,6 +22,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "time" "github.com/avast/retry-go/v4" @@ -157,7 +158,14 @@ func (q *offerQueue) retry(job offerJob) { select { case <-time.After(offerRetryInitialDelay): case <-ctx.Done(): - q.settle(job, ctx.Err()) + // No attempt has been made in this run yet, so ctx.Err() alone (e.g. "context deadline exceeded") + // carries no delivery-failure detail. Keep whatever real error a previous run already recorded + // (job.Error, persisted by OnRetry below) rather than overwriting it with a bare context error. + deadlineErr := ctx.Err() + if job.Error != "" { + deadlineErr = fmt.Errorf("%w (last recorded error: %s)", deadlineErr, job.Error) + } + q.settle(job, deadlineErr) return } @@ -171,6 +179,10 @@ func (q *offerQueue) retry(job offerJob) { retry.MaxJitter(offerRetryInitialDelay), retry.DelayType(retry.CombineDelay(retry.BackOffDelay, retry.RandomDelay)), retry.LastErrorOnly(true), + // Without this, retry.Do() returns a bare context error (e.g. "context deadline exceeded") when the + // offerRetryWindow deadline is hit between attempts, discarding the actual last delivery failure - + // exactly the detail a dead-lettered job's persisted Error should show an operator. + retry.WrapContextErrorWithLastError(true), retry.OnRetry(func(n uint, retryErr error) { job.Retries++ now := time.Now() diff --git a/vcr/issuer/offer_queue_test.go b/vcr/issuer/offer_queue_test.go index 8a4735931d..9cfecd9529 100644 --- a/vcr/issuer/offer_queue_test.go +++ b/vcr/issuer/offer_queue_test.go @@ -133,7 +133,7 @@ func TestOfferQueue_Schedule(t *testing.T) { q := newOfferQueue(db, func(_ context.Context, _ vc.VerifiableCredential) error { attempts.Add(1) - return errors.New("permanent failure") + return errors.New("permanent failure: wallet unreachable") }, func(_ context.Context, credential vc.VerifiableCredential) { givenUp <- credential @@ -158,6 +158,10 @@ func TestOfferQueue_Schedule(t *testing.T) { require.Len(t, failed, 1) require.True(t, failed[0].GivenUp) require.Equal(t, credential.ID.String(), failed[0].Credential.ID.String()) + // The persisted Error must show why delivery kept failing, not just that the window ran out: + // retry.Do() would otherwise discard the last real error in favor of a bare context error once + // the deadline is hit between attempts (see retry.WrapContextErrorWithLastError in offer_queue.go). + require.Contains(t, failed[0].Error, "wallet unreachable") }) t.Run("an unrecoverable error stops retrying immediately", func(t *testing.T) { From e728c1a85ed2474a8c55cfeae738be3707394b48 Mon Sep 17 00:00:00 2001 From: Rein Krul Date: Fri, 4 Sep 2026 14:50:30 +0200 Subject: [PATCH 5/6] fix(vcr): make offerQueue.Close() actually wait for in-flight retries Close() only cancelled the context and returned immediately, without waiting for retry goroutines to observe cancellation and stop. A caller closing the underlying KV store right after Close() returns (e.g. vcr.Shutdown() closing issuerStore/verifierStore/store right after issuer.Shutdown()) could race an in-flight save() still using context.Background() for persistence I/O. Track in-flight retries with a sync.WaitGroup and have Close() wait for them, bounded by offerQueueShutdownGrace (5s) so a misbehaving attempt that ignores cancellation can't hang shutdown forever - the same timeout-bounded-wait shape as e.g. http.Server.Shutdown(ctx), just with the bound owned internally rather than exposed on Shutdown()'s signature: core.Runnable (core/engine.go:205), which every module's Shutdown() implements, takes no context, so changing Issuer.Shutdown()'s signature would be a much larger, cross-cutting change out of scope here. Removed the sleep-based synchronization hack from the existing Close() test (no longer needed now that Close() genuinely blocks) and added tests for the actual blocking behavior and the grace-period timeout. Verified with -race -count=10, clean. Assisted by AI --- vcr/issuer/offer_queue.go | 36 +++++++++++++++-- vcr/issuer/offer_queue_test.go | 71 +++++++++++++++++++++++++++++++++- 2 files changed, 101 insertions(+), 6 deletions(-) diff --git a/vcr/issuer/offer_queue.go b/vcr/issuer/offer_queue.go index 210e8e1209..1b07823485 100644 --- a/vcr/issuer/offer_queue.go +++ b/vcr/issuer/offer_queue.go @@ -23,6 +23,7 @@ import ( "encoding/json" "errors" "fmt" + "sync" "time" "github.com/avast/retry-go/v4" @@ -72,6 +73,12 @@ type offerJob struct { GivenUp bool `json:"givenUp,omitempty"` } +// offerQueueShutdownGrace bounds how long Close() waits for in-flight retry goroutines to actually stop +// after being cancelled, before giving up on the wait and returning anyway. A well-behaved attempt (HTTP +// call using the per-job context) should stop almost immediately; this is a safety net against one that +// doesn't, so node shutdown can't hang forever on it. +var offerQueueShutdownGrace = 5 * time.Second + // offerQueue is a persistent, retrying queue for OpenID4VCI credential offers that failed on the initial // synchronous attempt. Modeled on network/dag's private-payload-fetch notifier: durable per-job state, // exponential backoff via retry-go, but bounded by a fixed total retry window rather than an attempt count. @@ -81,6 +88,7 @@ type offerQueue struct { giveUp offerGiveUpFn ctx context.Context cancel context.CancelFunc + wg sync.WaitGroup } // newOfferQueue creates an offerQueue backed by db. attempt is called for every (re)try; giveUp is called @@ -105,7 +113,7 @@ func (q *offerQueue) Schedule(credential vc.VerifiableCredential) error { if err := q.save(job); err != nil { return err } - go q.retry(job) + q.spawn(job) return nil } @@ -119,7 +127,7 @@ func (q *offerQueue) Run() error { if job.GivenUp { continue } - go q.retry(job) + q.spawn(job) } return nil } @@ -139,13 +147,33 @@ func (q *offerQueue) GetFailedOffers() ([]offerJob, error) { return failed, nil } -// Close stops all in-flight retries. Persisted jobs are left untouched; Run() picks them back up on the -// next startup. +// Close stops all in-flight retries and waits (up to offerQueueShutdownGrace) for them to actually return, +// so a caller closing the underlying store right after Close() returns doesn't race an in-flight save(). +// Persisted jobs are left untouched; Run() picks them back up on the next startup. func (q *offerQueue) Close() error { q.cancel() + stopped := make(chan struct{}) + go func() { + q.wg.Wait() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(offerQueueShutdownGrace): + log.Logger().Warn("Timed out waiting for OpenID4VCI offer retries to stop; some may still be running") + } return nil } +// spawn starts (or resumes) retrying job in the background, tracked by q.wg so Close() can wait for it. +func (q *offerQueue) spawn(job offerJob) { + q.wg.Add(1) + go func() { + defer q.wg.Done() + q.retry(job) + }() +} + func (q *offerQueue) retry(job offerJob) { deadline := job.FirstAttempt.Add(offerRetryWindow) ctx, cancel := context.WithDeadline(q.ctx, deadline) diff --git a/vcr/issuer/offer_queue_test.go b/vcr/issuer/offer_queue_test.go index 9cfecd9529..17747bca8f 100644 --- a/vcr/issuer/offer_queue_test.go +++ b/vcr/issuer/offer_queue_test.go @@ -268,12 +268,79 @@ func TestOfferQueue_Close(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("timed out waiting for first attempt") } - require.NoError(t, q.Close()) - time.Sleep(50 * time.Millisecond) // let the retry goroutine observe cancellation + require.NoError(t, q.Close()) // blocks until the retry goroutine has actually stopped jobs, err := q.all() require.NoError(t, err) require.Len(t, jobs, 1) require.False(t, jobs[0].GivenUp) }) + + t.Run("blocks until an in-flight attempt returns", func(t *testing.T) { + withFastRetryTiming(t, time.Second) + db := testOfferQueueStore(t) + inAttempt := make(chan struct{}) + releaseAttempt := make(chan struct{}) + var attemptReturned atomic.Bool + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + close(inAttempt) + <-releaseAttempt + attemptReturned.Store(true) + return errors.New("still failing") + }, + func(_ context.Context, _ vc.VerifiableCredential) { t.Fatal("giveUp should not be called on shutdown") }, + ) + + require.NoError(t, q.Schedule(testOfferQueueCredential(t, "did:nuts:issuer#8"))) + select { + case <-inAttempt: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the attempt to start") + } + + closeDone := make(chan struct{}) + go func() { + require.NoError(t, q.Close()) + close(closeDone) + }() + + select { + case <-closeDone: + t.Fatal("Close() returned before the in-flight attempt returned") + case <-time.After(100 * time.Millisecond): + } + + close(releaseAttempt) + select { + case <-closeDone: + case <-time.After(5 * time.Second): + t.Fatal("Close() did not return after the in-flight attempt returned") + } + require.True(t, attemptReturned.Load()) + }) + + t.Run("gives up waiting after the shutdown grace period", func(t *testing.T) { + withFastRetryTiming(t, time.Second) + original := offerQueueShutdownGrace + offerQueueShutdownGrace = 50 * time.Millisecond + t.Cleanup(func() { offerQueueShutdownGrace = original }) + + db := testOfferQueueStore(t) + stuck := make(chan struct{}) + q := newOfferQueue(db, + func(_ context.Context, _ vc.VerifiableCredential) error { + <-stuck // never returns on its own; ignores cancellation, like a misbehaving attempt would + return nil + }, + func(_ context.Context, _ vc.VerifiableCredential) {}, + ) + t.Cleanup(func() { close(stuck) }) + + require.NoError(t, q.Schedule(testOfferQueueCredential(t, "did:nuts:issuer#9"))) + + start := time.Now() + require.NoError(t, q.Close()) + require.Less(t, time.Since(start), time.Second, "Close() should have given up waiting after the grace period") + }) } From 4de481effddcb1568f5c62482b3f3d2cbc2e17af Mon Sep 17 00:00:00 2001 From: Rein Krul Date: Fri, 4 Sep 2026 14:58:35 +0200 Subject: [PATCH 6/6] docs(vcr): document offerJob fields, context.Background() rationale; log retries at Warn - Godoc on every offerJob field, including why the persisted key is Credential.ID.String() rather than a generated ID: idempotency (one credential can never have two persisted jobs) and it's the lookup key the planned admin requeue endpoint (#4469 item 3) needs anyway. - Expanded the context.Background()-vs-q.ctx comment on save/finish/all into its two actual reasons: reads must keep working independent of the queue's lifecycle (diagnostics), and writes must not be lost to the Close()/OnRetry shutdown race (q.ctx being cancelled mid-write would silently drop the last recorded Retries/Error state). - Bumped the per-retry-attempt log from Debug to Warn: every attempt that reaches the queue is a genuine delivery failure (unsupported or misconfigured wallet/issuer never reaches it - see issueUsingOpenID4VCI), so it deserves visibility, not Debug-level noise. Matches network/dag/notifier.go's OnRetry logging, the precedent this queue is modeled on, which logs every retry at Error/Warn rather than Debug. Assisted by AI --- vcr/issuer/offer_queue.go | 43 ++++++++++++++++++++++++++++++--------- 1 file changed, 33 insertions(+), 10 deletions(-) diff --git a/vcr/issuer/offer_queue.go b/vcr/issuer/offer_queue.go index 1b07823485..c28e98abaa 100644 --- a/vcr/issuer/offer_queue.go +++ b/vcr/issuer/offer_queue.go @@ -62,13 +62,25 @@ type offerGiveUpFn func(ctx context.Context, credential vc.VerifiableCredential) // issuer stopped supporting it between retries) and that retrying further won't help. var errOfferNoLongerSupported = errors.New("wallet or issuer no longer supports OpenID4VCI") -// offerJob is the persisted state of a single retrying credential offer. +// offerJob is the persisted state of a single retrying credential offer. It's stored keyed by +// Credential.ID.String() (see save/finish below), not a separate generated ID: that gives natural +// idempotency (the same VC can never end up with two persisted jobs) and is the lookup key the planned +// admin requeue endpoint (#4469 item 3, "requeue a stuck credential offer by credential ID") needs anyway. type offerJob struct { - Credential vc.VerifiableCredential `json:"credential"` - FirstAttempt time.Time `json:"firstAttempt"` - Retries int `json:"retries"` - Latest *time.Time `json:"latest,omitempty"` - Error string `json:"error,omitempty"` + // Credential is the offer being delivered. + Credential vc.VerifiableCredential `json:"credential"` + // FirstAttempt is when the offer was first scheduled. It's read from persisted state (not reset across + // restarts), since it anchors offerRetryWindow: the job is dead-lettered offerRetryWindow after this + // timestamp regardless of how many times the node has restarted in between. + FirstAttempt time.Time `json:"firstAttempt"` + // Retries counts failed attempts so far, incremented on every OnRetry callback. + Retries int `json:"retries"` + // Latest is when the job's state (Retries/Error, or GivenUp) was last updated. Nil until the first + // failed attempt. + Latest *time.Time `json:"latest,omitempty"` + // Error is the message from the most recent failed attempt, for diagnostics. Empty until the first + // failed attempt. + Error string `json:"error,omitempty"` // GivenUp indicates the retry window was exhausted; the offer is dead-lettered. GivenUp bool `json:"givenUp,omitempty"` } @@ -219,10 +231,15 @@ func (q *offerQueue) retry(job offerJob) { if saveErr := q.save(job); saveErr != nil { log.Logger().WithError(saveErr).Warn("Failed to persist OpenID4VCI offer retry state") } + // Warn, not Debug: every attempt reaching here is a genuine delivery failure (an unsupported or + // misconfigured wallet/issuer never reaches the queue at all - see issueUsingOpenID4VCI), so it's + // worth an operator's attention, not just Trace/Debug-level noise. Matches the give-up log level + // below and network/dag/notifier.go's equivalent OnRetry logging (the precedent this queue is + // modeled on), which logs every retry at Error/Warn rather than Debug. log.Logger(). WithError(retryErr). WithField(core.LogFieldCredentialID, job.Credential.ID.String()). - Debugf("Retrying OpenID4VCI credential offer (attempt %d)", n) + Warnf("Retrying OpenID4VCI credential offer (attempt %d)", n) }), ) q.settle(job, err) @@ -253,9 +270,15 @@ func (q *offerQueue) settle(job offerJob, err error) { q.giveUp(q.ctx, job.Credential) } -// Persistence operations deliberately use context.Background() rather than q.ctx: q.ctx is cancelled by -// Close() to stop in-flight retries, but reading/writing the persisted queue (e.g. from GetFailedOffers(), -// callable independently of whether the queue is still running) must keep working regardless. +// Persistence operations deliberately use context.Background() rather than q.ctx, for two reasons: +// 1. Reads must keep working regardless of whether the retry loop is running. GetFailedOffers() (and the +// admin requeue endpoint it'll back, #4469 item 3) is a diagnostics API with no reason to depend on the +// queue's own lifecycle - an operator should be able to inspect the DLQ even after Close(), or on an +// offerQueue that was never Run() at all. +// 2. Writes must not be lost to a shutdown race. save() is called from inside retry()'s OnRetry callback, +// which can fire in the narrow window where Close() has just cancelled q.ctx but the goroutine hasn't +// noticed yet. If save() used q.ctx, that write - the last real Retries/Error state before shutdown - +// would fail with "context canceled" and be silently dropped instead of persisted. func (q *offerQueue) save(job offerJob) error { data, err := json.Marshal(job)