pubsub: bump consensus, Corona→Corona field, fix raft
- Bump github.com/luxfi/consensus v1.22.84 → v1.25.17 (go mod tidy clean). - Migrate internal/consensus to current quasar API: QuasarHybridConsensus→Quasar, HybridSignature→QuasarSig, NewQuasarHybridConsensus→NewQuasar, VerifyHybridSignature→VerifyQuasarSig. Signature field is .Corona (no Corona). Route threshold<2 (single-node/dev) to NewTestQuasar, >=2 to NewQuasar (BFT), matching upstream's two-constructor production-vs-test split. - Repair botched upstream merge in server/: restore corrupted files from the upstream merge parent with Hanzo import paths preserved. Fixes RaftNodeCheckpoint redeclaration, undefined gsl/standalone/numLimitableConsumers, dropped consumer ack-format fields, resetStartingSeq arity, filestore IntersectGSL bool callbacks and fetchMsgNoCopyLocked, jetstream_cluster creatingRaftGroups/wca.State. go build ./... green; internal tests pass.
This commit is contained in:
zeekay
@@ -7,69 +7,85 @@ require (
|
||||
github.com/google/go-tpm v0.9.8
|
||||
github.com/hanzoai/pubsub-go v1.0.1-0.20260513042624-1b25bdfe16a6
|
||||
github.com/klauspost/compress v1.18.5
|
||||
github.com/luxfi/age v1.4.0
|
||||
github.com/luxfi/consensus v1.22.84
|
||||
github.com/luxfi/age v1.5.0
|
||||
github.com/luxfi/consensus v1.25.17
|
||||
github.com/luxfi/zap v0.8.1
|
||||
github.com/luxfi/zapdb v1.0.0
|
||||
github.com/luxfi/zapdb v1.10.0
|
||||
github.com/nats-io/jwt/v2 v2.8.1
|
||||
github.com/nats-io/nats-server/v2 v2.12.3
|
||||
github.com/nats-io/nats.go v1.50.0
|
||||
github.com/nats-io/nkeys v0.4.15
|
||||
github.com/nats-io/nuid v1.0.1
|
||||
github.com/zap-proto/go v0.2.0
|
||||
golang.org/x/crypto v0.52.0
|
||||
golang.org/x/sys v0.45.0
|
||||
golang.org/x/time v0.15.0
|
||||
)
|
||||
|
||||
require (
|
||||
filippo.io/edwards25519 v1.2.0 // indirect
|
||||
filippo.io/hpke v0.4.0 // indirect
|
||||
github.com/ALTree/bigfloat v0.2.0 // indirect
|
||||
github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20260311194731-d5b7577c683d // indirect
|
||||
github.com/bits-and-blooms/bitset v1.24.4 // indirect
|
||||
github.com/cenkalti/backoff v2.2.1+incompatible // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/cloudflare/circl v1.6.3 // indirect
|
||||
github.com/consensys/gnark-crypto v0.20.1 // indirect
|
||||
github.com/cronokirby/saferith v0.33.0 // indirect
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 // indirect
|
||||
github.com/dgraph-io/ristretto/v2 v2.4.0 // indirect
|
||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||
github.com/fxamacker/cbor/v2 v2.9.1 // indirect
|
||||
github.com/go-ini/ini v1.67.0 // indirect
|
||||
github.com/go-logr/logr v1.4.3 // indirect
|
||||
github.com/go-logr/stdr v1.2.2 // indirect
|
||||
github.com/google/flatbuffers v25.12.19+incompatible // indirect
|
||||
github.com/google/go-cmp v0.7.0 // indirect
|
||||
github.com/google/pprof v0.0.0-20260302011040-a15ffb7f9dcc // indirect
|
||||
github.com/google/uuid v1.6.0 // indirect
|
||||
github.com/gorilla/rpc v1.2.1 // indirect
|
||||
github.com/grandcat/zeroconf v1.0.0 // indirect
|
||||
github.com/gtank/merlin v0.1.1 // indirect
|
||||
github.com/gtank/ristretto255 v0.2.0 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
|
||||
github.com/klauspost/crc32 v1.3.0 // indirect
|
||||
github.com/luxfi/accel v1.1.9 // indirect
|
||||
github.com/luxfi/cache v1.2.1 // indirect
|
||||
github.com/luxfi/accel v1.2.2 // indirect
|
||||
github.com/luxfi/codec v1.1.4 // indirect
|
||||
github.com/luxfi/corona v0.7.6 // indirect
|
||||
github.com/luxfi/crypto v1.19.17 // indirect
|
||||
github.com/luxfi/crypto/ipa v1.2.4 // indirect
|
||||
github.com/luxfi/ids v1.2.9 // indirect
|
||||
github.com/luxfi/lattice/v7 v7.0.0 // indirect
|
||||
github.com/luxfi/ids v1.2.10 // indirect
|
||||
github.com/luxfi/lattice/v7 v7.1.4 // indirect
|
||||
github.com/luxfi/lens v0.1.4 // indirect
|
||||
github.com/luxfi/log v1.4.1 // indirect
|
||||
github.com/luxfi/magnetar v1.2.0 // indirect
|
||||
github.com/luxfi/math v1.4.1 // indirect
|
||||
github.com/luxfi/mdns v0.1.1 // indirect
|
||||
github.com/luxfi/metric v1.5.7 // indirect
|
||||
github.com/luxfi/mock v0.1.1 // indirect
|
||||
github.com/luxfi/corona v0.2.0 // indirect
|
||||
github.com/luxfi/utils v1.1.4 // indirect
|
||||
github.com/luxfi/pulsar v1.1.1 // indirect
|
||||
github.com/luxfi/threshold v1.9.4 // indirect
|
||||
github.com/mattn/go-colorable v0.1.14 // indirect
|
||||
github.com/mattn/go-isatty v0.0.20 // indirect
|
||||
github.com/miekg/dns v1.1.72 // indirect
|
||||
github.com/mimoo/StrobeGo v0.0.0-20220103164710-9a04d6ca976b // indirect
|
||||
github.com/minio/crc64nvme v1.1.1 // indirect
|
||||
github.com/minio/md5-simd v1.1.2 // indirect
|
||||
github.com/minio/minio-go/v7 v7.0.100 // indirect
|
||||
github.com/montanaflynn/stats v0.8.2 // indirect
|
||||
github.com/montanaflynn/stats v0.9.0 // indirect
|
||||
github.com/mr-tron/base58 v1.2.0 // indirect
|
||||
github.com/philhofer/fwd v1.2.0 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
|
||||
github.com/rs/xid v1.6.0 // indirect
|
||||
github.com/stretchr/testify v1.11.1 // indirect
|
||||
github.com/supranational/blst v0.3.16 // indirect
|
||||
github.com/tinylib/msgp v1.6.1 // indirect
|
||||
github.com/x448/float16 v0.8.4 // indirect
|
||||
github.com/zeebo/blake3 v0.2.4 // indirect
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
|
||||
go.opentelemetry.io/otel v1.40.0 // indirect
|
||||
go.opentelemetry.io/otel/metric v1.40.0 // indirect
|
||||
go.opentelemetry.io/otel/trace v1.40.0 // indirect
|
||||
go.opentelemetry.io/otel v1.43.0 // indirect
|
||||
go.opentelemetry.io/otel/metric v1.43.0 // indirect
|
||||
go.opentelemetry.io/otel/trace v1.43.0 // indirect
|
||||
go.uber.org/mock v0.6.0 // indirect
|
||||
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
||||
golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90 // indirect
|
||||
@@ -79,6 +95,7 @@ require (
|
||||
golang.org/x/text v0.37.0 // indirect
|
||||
golang.org/x/tools v0.45.0 // indirect
|
||||
google.golang.org/protobuf v1.36.11 // indirect
|
||||
gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
)
|
||||
|
||||
|
||||
@@ -1,45 +1,43 @@
|
||||
c2sp.org/CCTV/age v0.0.0-20251208015420-e9274a7bdbfd h1:ZLsPO6WdZ5zatV4UfVpr7oAwLGRZ+sebTUruuM4Ra3M=
|
||||
c2sp.org/CCTV/age v0.0.0-20251208015420-e9274a7bdbfd/go.mod h1:SrHC2C7r5GkDk8R+NFVzYy/sdj0Ypg9htaPXQq5Cqeo=
|
||||
filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
|
||||
filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
|
||||
filippo.io/hpke v0.4.0 h1:p575VVQ6ted4pL+it6M00V/f2qTZITO0zgmdKCkd5+A=
|
||||
filippo.io/hpke v0.4.0/go.mod h1:EmAN849/P3qdeK+PCMkDpDm83vRHM5cDipBJ8xbQLVY=
|
||||
github.com/ALTree/bigfloat v0.2.0 h1:AwNzawrpFuw55/YDVlcPw0F0cmmXrmngBHhVrvdXPvM=
|
||||
github.com/ALTree/bigfloat v0.2.0/go.mod h1:+NaH2gLeY6RPBPPQf4aRotPPStg+eXc8f9ZaE4vRfD4=
|
||||
github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20260215031811-a0ab0b218a81 h1:TBzelXBdnzDy+HCrBMcomEnhrmigkWOI1/mIPCi2u4M=
|
||||
github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20260215031811-a0ab0b218a81/go.mod h1:ioLG6R+5bUSO1oeGSDxOV3FADARuMoytZCSX6MEMQkI=
|
||||
github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20260311194731-d5b7577c683d/go.mod h1:ioLG6R+5bUSO1oeGSDxOV3FADARuMoytZCSX6MEMQkI=
|
||||
github.com/ChainSafe/go-schnorrkel v1.1.0 h1:rZ6EU+CZFCjB4sHUE1jIu8VDoB/wRKZxoe1tkcO71Wk=
|
||||
github.com/ChainSafe/go-schnorrkel v1.1.0/go.mod h1:ABkENxiP+cvjFiByMIZ9LYbRoNNLeBLiakC1XeTFxfE=
|
||||
github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
|
||||
github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0 h1:uWDG8BqLD1lI2ps38WDz2vXflrTX2+vLX0SvZtztJtE=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/bits-and-blooms/bitset v1.24.0 h1:H4x4TuulnokZKvHLfzVRTHJfFfnHEeSYJizujEZvmAM=
|
||||
github.com/bits-and-blooms/bitset v1.24.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
|
||||
github.com/bits-and-blooms/bitset v1.24.4 h1:95H15Og1clikBrKr/DuzMXkQzECs1M6hhoGXLwLQOZE=
|
||||
github.com/bits-and-blooms/bitset v1.24.4/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
|
||||
github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
|
||||
github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cloudflare/circl v1.6.1 h1:zqIqSPIndyBh1bjLVVDHMPpVKqp8Su/V+6MeDzzQBQ0=
|
||||
github.com/cloudflare/circl v1.6.1/go.mod h1:uddAzsPgqdMAYatqJ0lsjX1oECcQLIlRpzZh3pJrofs=
|
||||
github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8=
|
||||
github.com/cloudflare/circl v1.6.3/go.mod h1:2eXP6Qfat4O/Yhh8BznvKnJ+uzEoTQ6jVKJRn81BiS4=
|
||||
github.com/consensys/gnark-crypto v0.19.0 h1:zXCqeY2txSaMl6G5wFpZzMWJU9HPNh8qxPnYJ1BL9vA=
|
||||
github.com/consensys/gnark-crypto v0.19.0/go.mod h1:rT23F0XSZqE0mUA0+pRtnL56IbPxs6gp4CeRsBk4XS0=
|
||||
github.com/consensys/gnark-crypto v0.19.2 h1:qrEAIXq3T4egxqiliFFoNrepkIWVEeIYwt3UL0fvS80=
|
||||
github.com/consensys/gnark-crypto v0.19.2/go.mod h1:rT23F0XSZqE0mUA0+pRtnL56IbPxs6gp4CeRsBk4XS0=
|
||||
github.com/consensys/gnark-crypto v0.20.1 h1:PXDUBvk8AzhvWowHLWBEAfUQcV1/aZgWIqD6eMpXmDg=
|
||||
github.com/consensys/gnark-crypto v0.20.1/go.mod h1:RBWrSgy+IDbGR69RRV313th3M/aZU1ubk2om+qHuTSc=
|
||||
github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d h1:49RLWk1j44Xu4fjHb6JFYmeUnDORVwHNkDxaQ0ctCVU=
|
||||
github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d/go.mod h1:tSxLoYXyBmiFeKpvmq4dzayMdCjCnu8uqmCysIGBT2Y=
|
||||
github.com/cronokirby/saferith v0.33.0 h1:TgoQlfsD4LIwx71+ChfRcIpjkw+RPOapDEVxa+LhwLo=
|
||||
github.com/cronokirby/saferith v0.33.0/go.mod h1:QKJhjoqUtBsXCAVEjw38mFqoi7DebT7kthcD7UzbnoA=
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 h1:NMZiJj8QnKe1LgsbDayM4UoHwbvwDRwnI3hwNaAHRnc=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 h1:5RVFMOWjMyRy8cARdy79nAmgYw3hK/4HUq48LQ6Wwqo=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40=
|
||||
github.com/dgraph-io/ristretto/v2 v2.2.0 h1:bkY3XzJcXoMuELV8F+vS8kzNgicwQFAaGINAEJdWGOM=
|
||||
github.com/dgraph-io/ristretto/v2 v2.2.0/go.mod h1:RZrm63UmcBAaYWC1DotLYBmTvgkrs0+XhBd7Npn7/zI=
|
||||
github.com/dgraph-io/ristretto/v2 v2.4.0 h1:I/w09yLjhdcVD2QV192UJcq8dPBaAJb9pOuMyNy0XlU=
|
||||
github.com/dgraph-io/ristretto/v2 v2.4.0/go.mod h1:0KsrXtXvnv0EqnzyowllbVJB8yBonswa2lTCK2gGo9E=
|
||||
github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da h1:aIftn67I1fkbMa512G+w+Pxci9hJPB8oMnkcP3iZF38=
|
||||
github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
|
||||
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
|
||||
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
|
||||
github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ=
|
||||
github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
|
||||
github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A=
|
||||
github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
|
||||
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
|
||||
@@ -47,18 +45,26 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
|
||||
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
|
||||
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
|
||||
github.com/google/flatbuffers v25.2.10+incompatible h1:F3vclr7C3HpB1k9mxCGRMXq6FdUalZ6H/pNX4FP1v0Q=
|
||||
github.com/google/flatbuffers v25.2.10+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
|
||||
github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
|
||||
github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
|
||||
github.com/google/flatbuffers v25.12.19+incompatible h1:haMV2JRRJCe1998HeW/p0X9UaMTK6SDo0ffLn2+DbLs=
|
||||
github.com/google/flatbuffers v25.12.19+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
|
||||
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/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||
github.com/google/pprof v0.0.0-20260302011040-a15ffb7f9dcc h1:VBbFa1lDYWEeV5FZKUiYKYT0VxCp9twUmmaq9eb8sXw=
|
||||
github.com/google/pprof v0.0.0-20260302011040-a15ffb7f9dcc/go.mod h1:MxpfABSjhmINe3F1It9d+8exIHFvUqtLIRCdOGNXqiI=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/gorilla/rpc v1.2.1 h1:yC+LMV5esttgpVvNORL/xX4jvTTEUE30UZhZ5JF7K9k=
|
||||
github.com/gorilla/rpc v1.2.1/go.mod h1:uNpOihAlF5xRFLuTYhfR0yfCTm0WTQSQttkMSptRfGk=
|
||||
github.com/grandcat/zeroconf v1.0.0 h1:uHhahLBKqwWBV6WZUDAT71044vwOTL+McW0mBJvo6kE=
|
||||
github.com/grandcat/zeroconf v1.0.0/go.mod h1:lTKmG1zh86XyCoUeIHSA4FJMBwCJiQmGfcP2PdzytEs=
|
||||
github.com/gtank/merlin v0.1.1 h1:eQ90iG7K9pOhtereWsmyRJ6RAwcP4tHTDBHXNg+u5is=
|
||||
github.com/gtank/merlin v0.1.1/go.mod h1:T86dnYJhcGOh5BjZFCJWTDeTK7XW8uE+E21Cy/bIQ+s=
|
||||
github.com/gtank/ristretto255 v0.2.0 h1:LeOuWr6giplWkkMizx2emfG03SRPJqKt1nfIHLVHQ/0=
|
||||
github.com/gtank/ristretto255 v0.2.0/go.mod h1:OJ1ox/dWcp7sJ5grYDcZ+kkHYuj5nelW5aaL7ESVXBw=
|
||||
github.com/hanzoai/pubsub-go v1.0.1-0.20260513042624-1b25bdfe16a6 h1:3n1OhMquvkAb6jVFl3t819+dGOAK4QrX0aN+903OAbU=
|
||||
github.com/hanzoai/pubsub-go v1.0.1-0.20260513042624-1b25bdfe16a6/go.mod h1:18E+ko7z8kBNUlph5+KIAUFrqxt9PRo5BROlIO3AZxI=
|
||||
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
|
||||
@@ -68,61 +74,64 @@ github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzh
|
||||
github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
|
||||
github.com/klauspost/crc32 v1.3.0 h1:sSmTt3gUt81RP655XGZPElI0PelVTZ6YwCRnPSupoFM=
|
||||
github.com/klauspost/crc32 v1.3.0/go.mod h1:D7kQaZhnkX/Y0tstFGf8VUzv2UofNGqCjnC3zdHB0Hw=
|
||||
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
|
||||
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
||||
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
|
||||
github.com/leanovate/gopter v0.2.11 h1:vRjThO1EKPb/1NsDXuDrzldR28RLkBflWYcU9CvzWu4=
|
||||
github.com/leanovate/gopter v0.2.11/go.mod h1:aK3tzZP/C+p1m3SPRE4SYZFGP7jjkuSI4f7Xvpt0S9c=
|
||||
github.com/luxfi/accel v1.0.7 h1:ksHieAp50umwqxqgyHk9WiOmXM54kia3IEDb5H7FsM8=
|
||||
github.com/luxfi/accel v1.0.7/go.mod h1:iZD3oxffiMEIT/KvzD8bgwC/cBn4AYlMW3QJpbRa4RE=
|
||||
github.com/luxfi/accel v1.1.9 h1:Tsk6gXj2uKE19501bD0ajRYdeCHIlTGb6jYyLc+F8hc=
|
||||
github.com/luxfi/accel v1.1.9/go.mod h1:K00BcnLzEYMPHwCFq8Tf/450ApmTs9xBVvYOnofJCkc=
|
||||
github.com/luxfi/age v1.4.0 h1:tU5q65RQSQdaVq64Z/DVUhiPQMoMPQT6pr41LsvG0AQ=
|
||||
github.com/luxfi/age v1.4.0/go.mod h1:iAYAxgvrXxcy746+Ovh/eWWDuF9teJLNcCSSOX9RYW0=
|
||||
github.com/luxfi/cache v1.1.0 h1:6LUyGGZ+rrMAJBbAU6+UwkcamXj3zsboRUodIof2Ong=
|
||||
github.com/luxfi/cache v1.1.0/go.mod h1:9GvlEEE9rFPaaWxvVpSPwW8ZMo2+8VMNNcuPa4AwzPg=
|
||||
github.com/luxfi/cache v1.2.1/go.mod h1:co7JTxZZHpKT31Yh01LFp5aZOxmoUg157FhBLQdQHVU=
|
||||
github.com/luxfi/consensus v1.22.0 h1:zMUM8OUm+Pheuv9ln+/H/zK6Sm2TU9wUwqZiB4NKqlk=
|
||||
github.com/luxfi/consensus v1.22.0/go.mod h1:l8ZM8xMlljmq8YnIrJRm61Ed/bwxTsOY4B+tLsyjufY=
|
||||
github.com/luxfi/consensus v1.22.84 h1:fKw4n7fY+lCWW5HSx/ciGeRSxv4knODQGWh6lBWAveQ=
|
||||
github.com/luxfi/consensus v1.22.84/go.mod h1:y6y+bVjvpDLkkTBCCKLcch6R4PRW8EA+lTfr+6sB6F8=
|
||||
github.com/luxfi/crypto v1.17.4 h1:JcFaMC9EHjx/47XypQjl2woAysztl7a9kMsDY+pQh8I=
|
||||
github.com/luxfi/crypto v1.17.4/go.mod h1:Y+TkHDuHXe2vlwNlwkdI5TYds6C2d4OZcvmXjcMRZCQ=
|
||||
github.com/luxfi/crypto v1.19.0 h1:VtH6kvZrCEjCnkHPkhExDU+GJ0ZulYX4rnA+lXJdJHw=
|
||||
github.com/luxfi/crypto v1.19.0/go.mod h1:ee525i8Recbpb0jVTDZYZBr1MmvJ27OITJHZ/nlNMBw=
|
||||
github.com/luxfi/accel v1.2.2 h1:IzG1qUogapr4/KbbHP9avAimu3ioGJqh67WYKHxfZys=
|
||||
github.com/luxfi/accel v1.2.2/go.mod h1:5IvpNMfw9FgeBB9oobp+jbauAdhapfhc2bHYPUsHL9Y=
|
||||
github.com/luxfi/age v1.5.0 h1:G69HbSV4R3vKEH9B0CulnRaMdSdf4RalMgP8xKmxHeI=
|
||||
github.com/luxfi/age v1.5.0/go.mod h1:iAYAxgvrXxcy746+Ovh/eWWDuF9teJLNcCSSOX9RYW0=
|
||||
github.com/luxfi/codec v1.1.4 h1:Yl8ZalMNkqo7cD6R9AjczAajkLOmsjyZ9+DASVYHrvg=
|
||||
github.com/luxfi/codec v1.1.4/go.mod h1:oGQ3j6E8c2P0pL0irYtWkrB1hmDUFIE0puXHK4gV5KI=
|
||||
github.com/luxfi/consensus v1.25.17 h1:4EvtruOZSiInByIN9miha0SHlxGNlI5S59lhE1xa3LU=
|
||||
github.com/luxfi/consensus v1.25.17/go.mod h1:lvyDsuqWSDxwc0keI84PccZi8sRFu5zn+LAcPt79oIg=
|
||||
github.com/luxfi/corona v0.7.6 h1:CJP6smygD55dL0HHkKkWryL9H24a+wXvs+L+WchK7Nc=
|
||||
github.com/luxfi/corona v0.7.6/go.mod h1:4aD7+ZqnlZ2aVuU/DBQ5aspIagv5ux45LW2sJ4+siY8=
|
||||
github.com/luxfi/crypto v1.19.17 h1:l2LLu7UFyICtJVfraLDLRi+lFGiDXKHSL18M9/m1gsQ=
|
||||
github.com/luxfi/crypto v1.19.17/go.mod h1:INjdZtke85k8hX/QAmTMAY8bbZ4gzGZQLqURg3xf6Gk=
|
||||
github.com/luxfi/crypto/ipa v1.2.4 h1:6xfwhI9/HrcDkF3Ti5/NxsNQIWbwYDJmRSNIHRQ/xfU=
|
||||
github.com/luxfi/crypto/ipa v1.2.4/go.mod h1:43J6f6rcfUMrZt4cQectMOZb6Ps+fAEj8ZTPC3Kk+gE=
|
||||
github.com/luxfi/ids v1.1.2 h1:+qCUzE9Ga4slSHbnYl7T3I6c8y+n4/kKk4rzSEkLv/k=
|
||||
github.com/luxfi/ids v1.1.2/go.mod h1:cMbto3Q8N3IaBxdz4Lzaq9wYl33coGXUdlr2+YwXeUw=
|
||||
github.com/luxfi/ids v1.2.9 h1:+yjdhXW99drnd2Zlp1u/p8k3G23W3/1btJQ4ogHawUI=
|
||||
github.com/luxfi/ids v1.2.9/go.mod h1:khJOEdOPxd22yn0jcVrnbX1ADa0GHn5Y74gvCzN5BYc=
|
||||
github.com/luxfi/lattice/v7 v7.0.0 h1:d1vgan6mlb2KtwYfPc1g69uxVYaoVYZmlrIm4aCO88w=
|
||||
github.com/luxfi/lattice/v7 v7.0.0/go.mod h1:PFDdOkuGTQ0cbJMbKojzEJMGWUQmZW+wK9/wJ9F9fOs=
|
||||
github.com/luxfi/mdns v0.1.0 h1:VB3mQcETc9j5SY1S6lAgFtuGr/rjWuDgPYnxS+OKWMQ=
|
||||
github.com/luxfi/mdns v0.1.0/go.mod h1:/3dheKVjUk2yiS/ocH1IDzeLXOIe+kpVsErIGDOZdiQ=
|
||||
github.com/luxfi/ids v1.2.10 h1:f1WILZE199ayMuqnEyB2WP1qfMZkmozOQXSVYtB3e5k=
|
||||
github.com/luxfi/ids v1.2.10/go.mod h1:QBIwy3OHvrtskbUqKh1+OYRa6PsyR7f7oNX33sOfK7w=
|
||||
github.com/luxfi/lattice/v7 v7.1.4 h1:hQR02M6cHTAV5+joOPi9gb9Gm+z/hKJnhJF4IlciIJs=
|
||||
github.com/luxfi/lattice/v7 v7.1.4/go.mod h1:DmIQFi3mJiehVsR235l1NKYEU0JhU649OX5p7gMEW2c=
|
||||
github.com/luxfi/lens v0.1.4 h1:goGjGDXx2BNdjzXDunL5QT8elK2ZyCcc0z8TAbtWYrg=
|
||||
github.com/luxfi/lens v0.1.4/go.mod h1:mL+G8IK+9L41d78/2FYRgfhEzAjcr5+VEXB8SGuHbus=
|
||||
github.com/luxfi/log v1.4.1 h1:rIfFRodb9jrD/w7KayaUk0Oc+37PaQQdKEEMJCjR8gw=
|
||||
github.com/luxfi/log v1.4.1/go.mod h1:64IE3xRMJcpkQwnPUfJw3pDj7wU0kRS7BZ9wM7R72jk=
|
||||
github.com/luxfi/magnetar v1.2.0 h1:bsxHmBnJiswc/A6ElQ0pWz5g6ogqewIEKKqR26VgizA=
|
||||
github.com/luxfi/magnetar v1.2.0/go.mod h1:7J9YP9jByWbwCjssMFJNUkTU8tcPlSUoVSSiYShtvFs=
|
||||
github.com/luxfi/math v1.4.1 h1:1t9bCCsEqnl9yIKrShlbs80DBKyYTWdnzkVfBqEeO7Q=
|
||||
github.com/luxfi/math v1.4.1/go.mod h1:QvbRxauQyE1w4lvbcLSe6c8yeJz2Zj1Bq1rayGgs2tA=
|
||||
github.com/luxfi/mdns v0.1.1 h1:g2eRr9AXcziPkkcd24M+Qu9ApEpoKKjfI79QSNqv0rQ=
|
||||
github.com/luxfi/mdns v0.1.1/go.mod h1:dbp5f3h3aE7CGzwbaWzBM9cwdcekhmSrWhQevgYhhNA=
|
||||
github.com/luxfi/mock v0.1.0 h1:IwElfNu+T9sXvzFX6tudPDx1vqPuACRSRdxpD5lxW+o=
|
||||
github.com/luxfi/mock v0.1.0/go.mod h1:izF+9K0gGzFC9zERn6Po37v46eLdPB+EIsDjL3GLk+U=
|
||||
github.com/luxfi/metric v1.5.7 h1:LoSPEUpak2SLcynF+LT2cXjl9ECp4nY+Lia9zudmDv4=
|
||||
github.com/luxfi/metric v1.5.7/go.mod h1:CMguEhyuLi4YUWyXimJ+UHply99BDFrL0pxedB7rBqM=
|
||||
github.com/luxfi/mock v0.1.1 h1:0HEtIjg1J6CWz+IUyP6rsGqNWTcmxjFnSQIhaDuARwY=
|
||||
github.com/luxfi/mock v0.1.1/go.mod h1:jo35akl3Vtd8LbzDts8VJ0jmSVycrd1/eBi6g6t5hKU=
|
||||
github.com/luxfi/corona v0.2.0 h1:DbLMZmE//2T2wXXS/gEkKZrTbre9LD+7EE/tz5SQszU=
|
||||
github.com/luxfi/corona v0.2.0/go.mod h1:SZ+aDLUdfSAtaTRaaYzeDZ5DNQiLaOCelSaIGjL21R0=
|
||||
github.com/luxfi/utils v1.1.0 h1:ti7HvjNwJd4ILDMERJtOAWE9mF8l+zqDVkgWnF7Agic=
|
||||
github.com/luxfi/utils v1.1.0/go.mod h1:ABqhBdGNig0CaDcnNYldv1byS0BTNi5IKPbJSPF1p98=
|
||||
github.com/luxfi/utils v1.1.4/go.mod h1:c3yz1RjzrB+cs5GZm+q1T3/2cCKElO9vxm9yRRtgSEM=
|
||||
github.com/luxfi/zap v0.2.0 h1:RzvOkp3EoN5UCkpnqfObHLM1sEHy7YcxXKuermhE/VA=
|
||||
github.com/luxfi/zap v0.2.0/go.mod h1:2hydPSa/XMCMtfW6/DC9M5Bt03N5h75QwCV5Vypsqr4=
|
||||
github.com/luxfi/pq v1.0.3 h1:pFlQm1+5FuKTDUh2y/23bXWkN4I2Rc5iuxJypwDFFMs=
|
||||
github.com/luxfi/pq v1.0.3/go.mod h1:8bppZcRElfrVt0n3nYCZW3iX1TvhvzNbdjNdK1irgIE=
|
||||
github.com/luxfi/pulsar v1.1.1 h1:jo1jEgUsGiVxpT17Eg7Gw4Ax+07pbKFzq4NhHMLmDpI=
|
||||
github.com/luxfi/pulsar v1.1.1/go.mod h1:U7tPleeAHJ9dZ61ymtstzLKKoZjxM2zFeGZ+RSjHyRw=
|
||||
github.com/luxfi/threshold v1.9.4 h1:H69e9QkvygDtWkni1FkD5ztLdiTasmpJXcuVzgebdxo=
|
||||
github.com/luxfi/threshold v1.9.4/go.mod h1:o50rsc4H9fIXHuZXSf93qDFfRTXncXaXf1Q+6JAD1vU=
|
||||
github.com/luxfi/zap v0.8.1 h1:MiVRsRA808bXn6Orc81cjlGjqbn8gf7fr9aP5MPbLEM=
|
||||
github.com/luxfi/zap v0.8.1/go.mod h1:gUoQ/V9nTLLIb4/n0bY/L2zQDYK6pDBG121rK4t8bpg=
|
||||
github.com/luxfi/zapdb v1.0.0 h1:HFBlC8ka9RdP1MEXIGNNYg/0F6/Hu6O/OE6wKDtdk7g=
|
||||
github.com/luxfi/zapdb v1.0.0/go.mod h1:Qukh3hDRD0MnxA6z+a28JTnXhN85AiLLgp6TYr4QAMc=
|
||||
github.com/luxfi/zapdb/v4 v4.9.1 h1:oTg2ibBSxGtHTD9YHkclg5R1NecXWTi8R9XzY36uwgA=
|
||||
github.com/luxfi/zapdb/v4 v4.9.1/go.mod h1:gYJjLcFCxmUATZ6Zg7iZHwAbgBMxGVP1g8UOWomK70Y=
|
||||
github.com/luxfi/zapdb v1.10.0 h1:1lLHEmkyC0BucnA/zjQYsMkUVxuEo2vQkEaQGjYfuuc=
|
||||
github.com/luxfi/zapdb v1.10.0/go.mod h1:Qukh3hDRD0MnxA6z+a28JTnXhN85AiLLgp6TYr4QAMc=
|
||||
github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
|
||||
github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
|
||||
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
|
||||
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||
github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
|
||||
github.com/miekg/dns v1.1.62 h1:cN8OuEF1/x5Rq6Np+h1epln8OiyPWV+lROx9LxcGgIQ=
|
||||
github.com/miekg/dns v1.1.62/go.mod h1:mvDlcItzm+br7MToIKqkglaGhlFMHJ9DTNNWONWXbNQ=
|
||||
github.com/miekg/dns v1.1.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI=
|
||||
github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
|
||||
github.com/mimoo/StrobeGo v0.0.0-20181016162300-f8f6d4d2b643/go.mod h1:43+3pMjjKimDBf5Kr4ZFNGbLql1zKkbImw+fZbw3geM=
|
||||
github.com/mimoo/StrobeGo v0.0.0-20220103164710-9a04d6ca976b h1:QrHweqAtyJ9EwCaGHBu1fghwxIPiopAHV06JlXrMHjk=
|
||||
github.com/mimoo/StrobeGo v0.0.0-20220103164710-9a04d6ca976b/go.mod h1:xxLb2ip6sSUts3g1irPVHyk/DGslwQsNOo9I7smJfNU=
|
||||
github.com/minio/crc64nvme v1.1.1 h1:8dwx/Pz49suywbO+auHCBpCtlW1OfpcLN7wYgVR6wAI=
|
||||
github.com/minio/crc64nvme v1.1.1/go.mod h1:eVfm2fAzLlxMdUGc0EEBGSMmPwmXD5XiNRpnu9J3bvg=
|
||||
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
|
||||
@@ -131,8 +140,8 @@ github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
|
||||
github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
|
||||
github.com/minio/minio-go/v7 v7.0.100 h1:ShkWi8Tyj9RtU57OQB2HIXKz4bFgtVib0bbT1sbtLI8=
|
||||
github.com/minio/minio-go/v7 v7.0.100/go.mod h1:EtGNKtlX20iL2yaYnxEigaIvj0G0GwSDnifnG8ClIdw=
|
||||
github.com/montanaflynn/stats v0.8.2 h1:52wnefTJnPI5FoHif1DQh2soKRw0yYs+4AVyvtcZCH0=
|
||||
github.com/montanaflynn/stats v0.8.2/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
|
||||
github.com/montanaflynn/stats v0.9.0 h1:tsBJ0RXwph9BmAuFoCmqGv6e8xa0MENQ8m0ptKq29mQ=
|
||||
github.com/montanaflynn/stats v0.9.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
|
||||
github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
|
||||
github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
|
||||
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
|
||||
@@ -145,72 +154,71 @@ github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
|
||||
github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
github.com/onsi/ginkgo/v2 v2.28.1 h1:S4hj+HbZp40fNKuLUQOYLDgZLwNUVn19N3Atb98NCyI=
|
||||
github.com/onsi/ginkgo/v2 v2.28.1/go.mod h1:CLtbVInNckU3/+gC8LzkGUb9oF+e8W8TdUsxPwvdOgE=
|
||||
github.com/onsi/gomega v1.39.1 h1:1IJLAad4zjPn2PsnhH70V4DKRFlrCzGBNrNaru+Vf28=
|
||||
github.com/onsi/gomega v1.39.1/go.mod h1:hL6yVALoTOxeWudERyfppUcZXjMwIMLnuSfruD2lcfg=
|
||||
github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM=
|
||||
github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/quic-go/quic-go v0.59.1 h1:0Gmua0HW1Tv7ANR7hUYwRyD0MG5OJfgvYSZasGZzBic=
|
||||
github.com/quic-go/quic-go v0.59.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
|
||||
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
|
||||
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
|
||||
github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=
|
||||
github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
github.com/supranational/blst v0.3.16 h1:bTDadT+3fK497EvLdWRQEjiGnUtzJ7jjIUMF0jqwYhE=
|
||||
github.com/supranational/blst v0.3.16/go.mod h1:jZJtfjgudtNl4en1tzwPIV3KjUnQUvG3/j+w+fVonLw=
|
||||
github.com/tinylib/msgp v1.6.1 h1:ESRv8eL3u+DNHUoSAAQRE50Hm162zqAnBoGv9PzScPY=
|
||||
github.com/tinylib/msgp v1.6.1/go.mod h1:RSp0LW9oSxFut3KzESt5Voq4GVWyS+PSulT77roAqEA=
|
||||
github.com/zap-proto/go v0.2.0/go.mod h1:OITYSXRd0v93MWZuGl9XnPS77a09DYzN/Uy9/VwhuCs=
|
||||
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
|
||||
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
|
||||
github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
|
||||
github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
|
||||
github.com/zeebo/blake3 v0.2.4 h1:KYQPkhpRtcqh0ssGYcKLG1JYvddkEA8QwCM/yBqhaZI=
|
||||
github.com/zeebo/blake3 v0.2.4/go.mod h1:7eeQ6d2iXWRGF6npfaxl2CU+xy2Fjo2gxeyZGCRUjcE=
|
||||
go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
|
||||
go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
|
||||
github.com/zeebo/pcg v1.0.1 h1:lyqfGeWiv4ahac6ttHs+I5hwtH/+1mrhlCtVNQM2kHo=
|
||||
github.com/zeebo/pcg v1.0.1/go.mod h1:09F0S9iiKrwn9rlI5yjLkmrug154/YRW6KnnXVDM/l4=
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
|
||||
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
|
||||
go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
|
||||
go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I=
|
||||
go.opentelemetry.io/otel v1.40.0 h1:oA5YeOcpRTXq6NN7frwmwFR0Cn3RhTVZvXsP4duvCms=
|
||||
go.opentelemetry.io/otel v1.40.0/go.mod h1:IMb+uXZUKkMXdPddhwAHm6UfOwJyh4ct1ybIlV14J0g=
|
||||
go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE=
|
||||
go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
|
||||
go.opentelemetry.io/otel/metric v1.40.0 h1:rcZe317KPftE2rstWIBitCdVp89A2HqjkxR3c11+p9g=
|
||||
go.opentelemetry.io/otel/metric v1.40.0/go.mod h1:ib/crwQH7N3r5kfiBZQbwrTge743UDc7DTFVZrrXnqc=
|
||||
go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
|
||||
go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
|
||||
go.opentelemetry.io/otel/trace v1.40.0 h1:WA4etStDttCSYuhwvEa8OP8I5EWu24lkOzp+ZYblVjw=
|
||||
go.opentelemetry.io/otel/trace v1.40.0/go.mod h1:zeAhriXecNGP/s2SEG3+Y8X9ujcJOTqQ5RgdEJcawiA=
|
||||
go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
|
||||
go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0=
|
||||
go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM=
|
||||
go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY=
|
||||
go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A=
|
||||
go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
|
||||
go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
|
||||
go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
|
||||
go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
|
||||
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
|
||||
golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
|
||||
golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
|
||||
golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
|
||||
golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90 h1:jiDhWWeC7jfWqR9c/uplMOqJ0sbNlNWv0UkzE0vX1MA=
|
||||
golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90/go.mod h1:xE1HEv6b+1SCZ5/uscMRjUBKtIxworgEcEi+/n9NQDQ=
|
||||
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
|
||||
golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8=
|
||||
golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w=
|
||||
golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4=
|
||||
golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
|
||||
golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
|
||||
golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
|
||||
golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
|
||||
golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
|
||||
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
|
||||
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
|
||||
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
|
||||
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
@@ -219,12 +227,15 @@ golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
golang.org/x/tools v0.0.0-20191216052735-49a3e744a425/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
|
||||
golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
|
||||
golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
|
||||
golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8=
|
||||
golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0=
|
||||
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
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/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
|
||||
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
|
||||
gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
|
||||
gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
||||
@@ -24,11 +24,10 @@ import (
|
||||
"github.com/hanzoai/pubsub/internal/store"
|
||||
)
|
||||
|
||||
// Type aliases — the published v1.22.0 uses legacy "Hybrid" names.
|
||||
// We alias to the correct names here; a new release will export these directly.
|
||||
// Type aliases to the upstream consensus quasar primitives.
|
||||
type (
|
||||
QuasarConsensus = quasar.QuasarHybridConsensus
|
||||
QuasarSignature = quasar.HybridSignature
|
||||
QuasarConsensus = quasar.Quasar
|
||||
QuasarSignature = quasar.QuasarSig
|
||||
)
|
||||
|
||||
// maxPendingOps caps the pending map size to prevent OOM from proposal spam.
|
||||
@@ -170,13 +169,19 @@ func New(cfg Config) (*Quasar, error) {
|
||||
cfg.ZAPPort = 0
|
||||
}
|
||||
|
||||
signer, err := quasar.NewQuasarHybridConsensus(cfg.Threshold)
|
||||
// Production BFT consensus requires threshold >= 2 (NewQuasar). A threshold
|
||||
// of 1 is a single-node/dev topology, which upstream exposes via NewTestQuasar.
|
||||
newSigner := quasar.NewQuasar
|
||||
if cfg.Threshold < 2 {
|
||||
newSigner = quasar.NewTestQuasar
|
||||
}
|
||||
signer, err := newSigner(cfg.Threshold)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("quasar: init signer: %w", err)
|
||||
}
|
||||
|
||||
for _, id := range cfg.ValidatorIDs {
|
||||
if err := signer.AddValidator(id, 1); err != nil {
|
||||
if _, err := signer.AddValidator(id, 1); err != nil {
|
||||
return nil, fmt.Errorf("quasar: add validator %s: %w", id, err)
|
||||
}
|
||||
}
|
||||
@@ -315,7 +320,7 @@ func (q *Quasar) addVoteLocked(hash string, validatorID string, sig *QuasarSigna
|
||||
if err != nil || len(hashBytes) == 0 {
|
||||
return false
|
||||
}
|
||||
if !q.signer.VerifyHybridSignature(hashBytes, sig) {
|
||||
if !q.signer.VerifyQuasarSig(hashBytes, sig) {
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
+36
-12
@@ -442,6 +442,9 @@ type consumer struct {
|
||||
rlimit *rate.Limiter
|
||||
reqSub *subscription
|
||||
resetSub *subscription
|
||||
ackSubOld *subscription
|
||||
ackReplyOldT string
|
||||
ackSubjOld string
|
||||
ackSub *subscription
|
||||
ackReplyT string
|
||||
ackSubj string
|
||||
@@ -1111,10 +1114,12 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
|
||||
return nil, NewJSConsumerDoesNotExistError()
|
||||
}
|
||||
|
||||
standalone := !s.JetStreamIsClustered() && s.standAloneMode()
|
||||
|
||||
// If we're clustered we've already done this check, only do this if we're a standalone server.
|
||||
// But if we're standalone, only enforce if we're not recovering, since the MaxConsumers could've
|
||||
// been updated while we already had more consumers on disk.
|
||||
if !s.JetStreamIsClustered() && s.standAloneMode() && !isRecovering {
|
||||
if standalone && !isRecovering {
|
||||
// Check for any limits, if the config for the consumer sets a limit we check against that
|
||||
// but if not we use the value from account limits, if account limits is more restrictive
|
||||
// than stream config we prefer the account limits to handle cases where account limits are
|
||||
@@ -1123,7 +1128,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
|
||||
if maxc <= 0 || (selectedLimits.MaxConsumers > 0 && selectedLimits.MaxConsumers < maxc) {
|
||||
maxc = selectedLimits.MaxConsumers
|
||||
}
|
||||
if maxc > 0 && mset.numPublicConsumers() >= maxc {
|
||||
if maxc > 0 && mset.numLimitableConsumers() >= maxc {
|
||||
mset.mu.Unlock()
|
||||
return nil, NewJSMaximumConsumersLimitError()
|
||||
}
|
||||
@@ -1714,10 +1719,6 @@ func (o *consumer) setLeader(isLeader bool) error {
|
||||
o.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
if o.resetSub, err = o.subscribeInternal(o.resetSubj, o.processResetReq); err != nil {
|
||||
o.mu.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
// Check on flow control settings.
|
||||
if o.cfg.FlowControl {
|
||||
@@ -1846,8 +1847,9 @@ func (o *consumer) setLeader(isLeader bool) error {
|
||||
o.unsubscribe(o.ackSub)
|
||||
o.unsubscribe(o.reqSub)
|
||||
o.unsubscribe(o.resetSub)
|
||||
o.unsubscribe(o.fcSubOld)
|
||||
o.unsubscribe(o.fcSub)
|
||||
o.ackSub, o.reqSub, o.resetSub, o.fcSub = nil, nil, nil, nil
|
||||
o.ackSubOld, o.ackSub, o.reqSub, o.resetSub, o.fcSubOld, o.fcSub = nil, nil, nil, nil, nil, nil
|
||||
if o.infoSub != nil {
|
||||
o.srv.sysUnsubscribe(o.infoSub)
|
||||
o.infoSub = nil
|
||||
@@ -2627,6 +2629,9 @@ func (o *consumer) updateConfig(cfg *ConsumerConfig) error {
|
||||
// Allowed but considered no-op, [Description, SampleFrequency, MaxWaiting, HeadersOnly]
|
||||
o.cfg = *cfg
|
||||
|
||||
if cfg.Sourcing && (!o.srv.JetStreamIsClustered() && o.srv.standAloneMode()) {
|
||||
o.resetStartingSeqLocked(0, _EMPTY_, false)
|
||||
}
|
||||
if updatedFilters {
|
||||
// Cleanup messages that lost interest.
|
||||
if o.retention == InterestPolicy {
|
||||
@@ -2792,10 +2797,14 @@ func (o *consumer) updateSkipped(seq uint64) {
|
||||
o.propose(b[:])
|
||||
}
|
||||
|
||||
func (o *consumer) resetStartingSeq(seq uint64, reply string) (uint64, bool, error) {
|
||||
func (o *consumer) resetStartingSeq(seq uint64, reply string, internal bool) (uint64, bool, error) {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
return o.resetStartingSeqLocked(seq, reply, internal)
|
||||
}
|
||||
|
||||
// Lock should be held.
|
||||
func (o *consumer) resetStartingSeqLocked(seq uint64, reply string, internal bool) (uint64, bool, error) {
|
||||
// Reset to a specific sequence, or back to the ack floor.
|
||||
if seq == 0 {
|
||||
seq = o.asflr + 1
|
||||
@@ -2824,17 +2833,27 @@ VALID:
|
||||
if seq <= 0 {
|
||||
seq = 1
|
||||
}
|
||||
o.resetLocalStartingSeq(seq)
|
||||
// Clustered mode and R>1.
|
||||
// The replicated path requires quorum first before the reset actually takes effect.
|
||||
if o.node != nil {
|
||||
if !o.isLeader() {
|
||||
return 0, false, nil
|
||||
}
|
||||
b := make([]byte, 1+8+len(reply))
|
||||
b[0] = byte(resetSeqOp)
|
||||
var le = binary.LittleEndian
|
||||
le.PutUint64(b[1:], seq)
|
||||
copy(b[1+8:], reply)
|
||||
o.propose(b[:])
|
||||
if reply != _EMPTY_ {
|
||||
if o.rsm == nil {
|
||||
o.rsm = make(map[string]bool, 1)
|
||||
}
|
||||
o.rsm[reply] = internal
|
||||
}
|
||||
return seq, false, nil
|
||||
} else if o.store != nil {
|
||||
}
|
||||
o.resetLocalStartingSeq(seq)
|
||||
if o.store != nil {
|
||||
o.store.Reset(seq - 1)
|
||||
// Cleanup messages that lost interest.
|
||||
if o.retention == InterestPolicy {
|
||||
@@ -4498,7 +4517,7 @@ func (o *consumer) processResetReq(_ *subscription, c *client, a *Account, _, re
|
||||
return
|
||||
}
|
||||
}
|
||||
resetSeq, canRespond, err := o.resetStartingSeq(req.Seq, reply)
|
||||
resetSeq, canRespond, err := o.resetStartingSeq(req.Seq, reply, false)
|
||||
if err != nil {
|
||||
resp.Error = NewJSConsumerInvalidResetError(err)
|
||||
s.sendInternalAccountMsg(a, reply, s.jsonResponse(&resp))
|
||||
@@ -4813,6 +4832,9 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
|
||||
// That will correct the pending state and delivery/ack floors, so just skip here.
|
||||
pmsg.returnToPool()
|
||||
pmsg = nil
|
||||
if p, ok := o.pending[seq]; ok {
|
||||
o.processTermLocked(seq, p.Sequence, dc-1, ackTermUnackedLimitsReason, _EMPTY_, false)
|
||||
}
|
||||
continue
|
||||
}
|
||||
return pmsg, dc, err
|
||||
@@ -6583,11 +6605,13 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
|
||||
o.unsubscribe(o.ackSub)
|
||||
o.unsubscribe(o.reqSub)
|
||||
o.unsubscribe(o.resetSub)
|
||||
o.unsubscribe(o.fcSubOld)
|
||||
o.unsubscribe(o.fcSub)
|
||||
o.ackSubOld = nil
|
||||
o.ackSub = nil
|
||||
o.reqSub = nil
|
||||
o.resetSub = nil
|
||||
o.fcSubOld = nil
|
||||
o.fcSub = nil
|
||||
if o.infoSub != nil {
|
||||
o.srv.sysUnsubscribe(o.infoSub)
|
||||
|
||||
+108
-57
@@ -1212,8 +1212,8 @@ func (fs *fileStore) recoverMsgBlock(index uint32) (*msgBlock, error) {
|
||||
}
|
||||
// We can recycle it now.
|
||||
recycleMsgBlockBuf(buf)
|
||||
} else {
|
||||
file.ReadAt(lchk[:], int64(mb.rbytes)-checksumSize)
|
||||
} else if _, err = file.ReadAt(lchk[:], int64(mb.rbytes)-checksumSize); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1686,8 +1686,13 @@ func (mb *msgBlock) rebuildStateFromBufLocked(buf []byte, allowTruncate bool) (*
|
||||
dlen := int(rl) - msgHdrSize
|
||||
// Do some quick sanity checks here.
|
||||
if dlen < 0 || shlen > (dlen-recordHashSize) || dlen > int(rl) || index+rl > lbuf || rl > rlBadThresh {
|
||||
truncate(index)
|
||||
return gatherLost(lbuf - index), tombstones, errBadMsg{mb.mfn, fmt.Sprintf("sanity check failed (dlen %d slen %d rl %d index %d lbuf %d)", dlen, slen, rl, index, lbuf)}
|
||||
err = errBadMsg{mb.mfn, fmt.Sprintf("sanity check failed (dlen %d slen %d rl %d index %d lbuf %d)", dlen, slen, rl, index, lbuf)}
|
||||
if allowTruncate {
|
||||
if err = truncate(index); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
return gatherLost(lbuf - index), tombstones, err
|
||||
}
|
||||
|
||||
// Check for checksum failures before additional processing.
|
||||
@@ -2431,7 +2436,10 @@ func (fs *fileStore) recoverMsgs() error {
|
||||
// out from underneath of us this is possible.
|
||||
mb.mu.Lock()
|
||||
if atomic.LoadUint64(&mb.first.seq) == 0 {
|
||||
mb.dirtyCloseWithRemove(true)
|
||||
if err := mb.dirtyCloseWithRemove(true); err != nil {
|
||||
mb.mu.Unlock()
|
||||
return err
|
||||
}
|
||||
fs.removeMsgBlockFromList(mb)
|
||||
mb.mu.Unlock()
|
||||
continue
|
||||
@@ -2769,6 +2777,7 @@ func (fs *fileStore) GetSeqFromTime(t time.Time) uint64 {
|
||||
|
||||
// Using a binary search, but need to be aware of interior deletes in the block.
|
||||
seq := lseq + 1
|
||||
mb.mu.Lock()
|
||||
loop:
|
||||
for fseq <= lseq {
|
||||
mid := fseq + (lseq-fseq)/2
|
||||
@@ -2776,7 +2785,7 @@ loop:
|
||||
// Potentially skip over gaps. We keep the original middle but keep track of a
|
||||
// potential delete range with an offset.
|
||||
for {
|
||||
sm, _, err := mb.fetchMsgNoCopy(mid+off, &smv)
|
||||
sm, _, err := mb.fetchMsgNoCopyLocked(mid+off, &smv)
|
||||
if err != nil || sm == nil {
|
||||
off++
|
||||
if mid+off <= lseq {
|
||||
@@ -2803,6 +2812,8 @@ loop:
|
||||
fseq = mid + off + 1
|
||||
}
|
||||
}
|
||||
mb.finishedWithCache()
|
||||
mb.mu.Unlock()
|
||||
return seq
|
||||
}
|
||||
|
||||
@@ -2843,10 +2854,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *gsl.SimpleSublist, start uint64, sm *
|
||||
// If there are no subject matches then this is effectively no-op.
|
||||
hseq := uint64(math.MaxUint64)
|
||||
var ierr error
|
||||
stree.IntersectGSL(mb.fss, sl, func(subj []byte, ss *SimpleState) {
|
||||
if ierr != nil {
|
||||
return
|
||||
}
|
||||
stree.IntersectGSL(mb.fss, sl, func(subj []byte, ss *SimpleState) bool {
|
||||
if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
|
||||
// mb is already loaded into the cache so should be fast-ish.
|
||||
if ierr = mb.recalculateForSubj(bytesToString(subj), ss); ierr != nil {
|
||||
@@ -2870,16 +2878,6 @@ func (mb *msgBlock) firstMatchingMulti(sl *gsl.SimpleSublist, start uint64, sm *
|
||||
if sm == nil {
|
||||
sm = new(StoreMsg)
|
||||
}
|
||||
// Need messages loaded from here on out.
|
||||
if mb.cacheNotLoaded() {
|
||||
if ierr = mb.loadMsgsWithLock(); ierr != nil {
|
||||
return
|
||||
}
|
||||
didLoad = true
|
||||
}
|
||||
if sm == nil {
|
||||
sm = new(StoreMsg)
|
||||
}
|
||||
if first == ss.First {
|
||||
// If the start floor is below where this subject starts then we can
|
||||
// short-circuit, avoiding needing to scan for the next message.
|
||||
@@ -3464,16 +3462,29 @@ func (fs *fileStore) checkSkipFirstBlock(filter string, wc bool, bi int) (int, e
|
||||
// This is used to see if we can selectively jump start blocks based on filter subjects and a starting block index.
|
||||
// Will return -1 and ErrStoreEOF if no matches at all or no more from where we are.
|
||||
func (fs *fileStore) checkSkipFirstBlockMulti(sl *gsl.SimpleSublist, bi int) (int, error) {
|
||||
// Don't bother if full wildcard.
|
||||
if sl.MatchesFullWildcard() || fs.psim.Size() > highCardinalityThreshold {
|
||||
return bi + 1, nil
|
||||
}
|
||||
// Move through psim to gather start and stop bounds.
|
||||
start, stop := uint32(math.MaxUint32), uint32(0)
|
||||
stree.IntersectGSL(fs.psim, sl, func(subj []byte, psi *psi) {
|
||||
guard := fs.blks[bi].getIndex() + 1
|
||||
stree.IntersectGSL(fs.psim, sl, func(subj []byte, psi *psi) bool {
|
||||
if psi.fblk < start {
|
||||
start = psi.fblk
|
||||
}
|
||||
if start == guard {
|
||||
// One of the subjects matches the next block, so there's no point in carrying on trying to skip.
|
||||
return false
|
||||
}
|
||||
if psi.lblk > stop {
|
||||
stop = psi.lblk
|
||||
}
|
||||
return true
|
||||
})
|
||||
if start == guard {
|
||||
return bi + 1, nil
|
||||
}
|
||||
// Nothing was found.
|
||||
if start == uint32(math.MaxUint32) {
|
||||
return -1, ErrStoreEOF
|
||||
@@ -3748,7 +3759,9 @@ func (fs *fileStore) allLastSeqsLocked() ([]uint64, error) {
|
||||
// Check if we need to recalculate. We only care about the last sequence.
|
||||
if ss.lastNeedsUpdate {
|
||||
// mb is already loaded into the cache so should be fast-ish.
|
||||
mb.recalculateForSubj(bytesToString(bsubj), ss)
|
||||
if ierr = mb.recalculateForSubj(bytesToString(bsubj), ss); ierr != nil {
|
||||
return false
|
||||
}
|
||||
}
|
||||
seqs = append(seqs, ss.Last)
|
||||
subs[string(bsubj)] = struct{}{}
|
||||
@@ -5380,8 +5393,9 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
|
||||
|
||||
bsubj := stringToBytes(subj)
|
||||
if ss, ok := mb.fss.Find(bsubj); ok && ss != nil {
|
||||
var err error
|
||||
if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
|
||||
mb.recalculateForSubj(subj, ss)
|
||||
err = mb.recalculateForSubj(subj, ss)
|
||||
}
|
||||
mb.mu.Unlock()
|
||||
// Re-acquire fs lock
|
||||
@@ -5392,6 +5406,9 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
|
||||
info.fblk = i
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return ss.First, nil
|
||||
}
|
||||
// If we did not find it and we loaded this msgBlock try to expire as long as not the last.
|
||||
@@ -5709,7 +5726,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
|
||||
|
||||
// Remove a message from the given block, optionally rewriting the mb file.
|
||||
// fs lock should be held.
|
||||
func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLimits bool) (bool, error) {
|
||||
func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLimits bool) (removed bool, rerr error) {
|
||||
mb.mu.Lock()
|
||||
|
||||
// See if we are closed or the sequence number is still relevant or if we know its deleted.
|
||||
@@ -5753,7 +5770,6 @@ func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLim
|
||||
lhdr, lmsg int
|
||||
ttl int64
|
||||
)
|
||||
// We don't use a copy as long as that's possible. When unlocking mb or erasing, we'll copy the subject.
|
||||
sm, err := mb.cacheLookupNoCopy(seq, &smv)
|
||||
if err != nil {
|
||||
finishedWithCache()
|
||||
@@ -5764,7 +5780,9 @@ func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLim
|
||||
}
|
||||
return false, err
|
||||
} else if sm != nil {
|
||||
subj = sm.subj
|
||||
// subj aliases mb.cache.buf; copy now because the cache may be erased or
|
||||
// recycled after we drop mb.mu. The rest are scalars stashed for later use.
|
||||
subj = copyString(sm.subj)
|
||||
ts = sm.ts
|
||||
lhdr = len(sm.hdr)
|
||||
lmsg = len(sm.msg)
|
||||
@@ -5776,17 +5794,18 @@ func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLim
|
||||
// when the last block is empty.
|
||||
// If not via limits and not empty (empty writes tombstone below if last) write tombstone.
|
||||
if !viaLimits && !isEmpty && sm != nil {
|
||||
// Need to copy the subject since we unlock and re-acquire, and the cache could change.
|
||||
subj = copyString(subj)
|
||||
mb.mu.Unlock() // Only safe way to checkLastBlock is to unlock here...
|
||||
lmb, err := fs.checkLastBlock(emptyRecordLen)
|
||||
if err != nil {
|
||||
mb.mu.Lock()
|
||||
finishedWithCache()
|
||||
mb.mu.Unlock()
|
||||
return false, err
|
||||
}
|
||||
if err := lmb.writeTombstone(seq, ts); err != nil {
|
||||
mb.mu.Lock()
|
||||
finishedWithCache()
|
||||
mb.mu.Unlock()
|
||||
return false, err
|
||||
}
|
||||
mb.mu.Lock() // We'll need the lock back to carry on safely.
|
||||
@@ -5807,9 +5826,6 @@ func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLim
|
||||
mb.mu.Unlock()
|
||||
return false, err
|
||||
}
|
||||
// Need to copy the subject, as eraseMsg will overwrite the cache and we won't
|
||||
// be able to access sm.subj anymore later on.
|
||||
subj = copyString(subj)
|
||||
if err := mb.eraseMsg(seq, int(ri), int(msz), isLastBlock); err != nil {
|
||||
finishedWithCache()
|
||||
mb.mu.Unlock()
|
||||
@@ -5851,7 +5867,11 @@ func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLim
|
||||
}
|
||||
|
||||
// If we are tracking multiple subjects here make sure we update that accounting.
|
||||
mb.removeSeqPerSubject(subj, seq)
|
||||
if _, err = mb.removeSeqPerSubject(subj, seq); err != nil {
|
||||
finishedWithCache()
|
||||
mb.mu.Unlock()
|
||||
return false, err
|
||||
}
|
||||
fs.removePerSubject(subj)
|
||||
if fs.ttls != nil && ttl > 0 {
|
||||
expires := time.Duration(ts) + (time.Second * time.Duration(ttl))
|
||||
@@ -5933,18 +5953,17 @@ func (fs *fileStore) removeMsgFromBlock(mb *msgBlock, seq uint64, secure, viaLim
|
||||
|
||||
// Remove all messages in the range [first, last]
|
||||
// Lock should be held.
|
||||
func (fs *fileStore) removeMsgsInRange(first, last uint64, viaLimits bool) {
|
||||
func (fs *fileStore) removeMsgsInRange(first, last uint64, viaLimits bool) error {
|
||||
last = min(last, fs.state.LastSeq)
|
||||
if first > last {
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
firstBlock := sort.Search(len(fs.blks), func(i int) bool {
|
||||
return atomic.LoadUint64(&fs.blks[i].last.seq) >= first
|
||||
})
|
||||
|
||||
if firstBlock >= len(fs.blks) {
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
for i := firstBlock; i < len(fs.blks); {
|
||||
@@ -5961,16 +5980,21 @@ func (fs *fileStore) removeMsgsInRange(first, last uint64, viaLimits bool) {
|
||||
// purgeMgsBlock, which also removes the block from fs.blks.
|
||||
// After purgeMsgBlock, i will be the index of the following
|
||||
// msgBlock, if any. Therefore, continue without incrementing i.
|
||||
fs.purgeMsgBlock(mb)
|
||||
if err := fs.purgeMsgBlock(mb); err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
from := max(first, mbFirstSeq)
|
||||
to := min(last, mbLastSeq)
|
||||
for seq := from; seq <= to; seq++ {
|
||||
fs.removeMsgFromBlock(mb, seq, false, viaLimits)
|
||||
if _, err := fs.removeMsgFromBlock(mb, seq, false, viaLimits); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
i++
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Tests whether we should try to compact this block while inline removing msgs.
|
||||
@@ -8896,9 +8920,17 @@ func (fs *fileStore) SubjectForSeq(seq uint64) (string, error) {
|
||||
mb := fs.selectMsgBlock(seq)
|
||||
fs.mu.RUnlock()
|
||||
if mb != nil {
|
||||
if sm, _, _ := mb.fetchMsgNoCopy(seq, &smv); sm != nil {
|
||||
mb.mu.Lock()
|
||||
sm, _, _ := mb.fetchMsgNoCopyLocked(seq, &smv)
|
||||
var subj string
|
||||
if sm != nil {
|
||||
// Copy the subject, as it's used elsewhere, and the backing cache could be reused in the meantime.
|
||||
return copyString(sm.subj), nil
|
||||
subj = copyString(sm.subj)
|
||||
}
|
||||
mb.finishedWithCache()
|
||||
mb.mu.Unlock()
|
||||
if sm != nil {
|
||||
return subj, nil
|
||||
}
|
||||
}
|
||||
return _EMPTY_, ErrStoreMsgNotFound
|
||||
@@ -8978,7 +9010,10 @@ func (fs *fileStore) loadLastLocked(subj string, sm *StoreMsg) (lsm *StoreMsg, e
|
||||
// Check if we need to recalculate. We only care about the last sequence.
|
||||
if ss.lastNeedsUpdate {
|
||||
// mb is already loaded into the cache so should be fast-ish.
|
||||
mb.recalculateForSubj(subj, ss)
|
||||
if err = mb.recalculateForSubj(subj, ss); err != nil {
|
||||
mb.mu.Unlock()
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
l = ss.Last
|
||||
}
|
||||
@@ -9052,12 +9087,13 @@ func (fs *fileStore) LoadNextMsgMulti(sl *gsl.SimpleSublist, start uint64, smp *
|
||||
if start <= fs.state.FirstSeq {
|
||||
var total uint64
|
||||
blkStart := uint32(math.MaxUint32)
|
||||
stree.IntersectGSL(fs.psim, sl, func(subj []byte, psi *psi) {
|
||||
stree.IntersectGSL(fs.psim, sl, func(subj []byte, psi *psi) bool {
|
||||
total += psi.total
|
||||
// Keep track of start index for this subject.
|
||||
if psi.fblk < blkStart {
|
||||
blkStart = psi.fblk
|
||||
}
|
||||
return true
|
||||
})
|
||||
// Nothing available.
|
||||
if total == 0 {
|
||||
@@ -9980,7 +10016,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
|
||||
// When writing multiple tombstones we will flush at the end.
|
||||
if len(tombs) > 0 {
|
||||
for _, tomb := range tombs {
|
||||
if err := fs.writeTombstoneNoFlush(tomb.seq, tomb.ts); err != nil {
|
||||
if err = fs.writeTombstoneNoFlush(tomb.seq, tomb.ts); err != nil {
|
||||
fs.mu.Unlock()
|
||||
return purged, err
|
||||
}
|
||||
@@ -10478,7 +10514,7 @@ SKIP:
|
||||
// When writing multiple tombstones we will flush at the end.
|
||||
if len(tombs) > 0 {
|
||||
for _, tomb := range tombs {
|
||||
if err := fs.writeTombstoneNoFlush(tomb.seq, tomb.ts); err != nil {
|
||||
if err = fs.writeTombstoneNoFlush(tomb.seq, tomb.ts); err != nil {
|
||||
fs.mu.Unlock()
|
||||
return purged, err
|
||||
}
|
||||
@@ -11699,7 +11735,7 @@ func (fs *fileStore) _writeFullState(force bool) error {
|
||||
if fs.state.LastSeq > fs.state.FirstSeq {
|
||||
numDeleted = int((fs.state.LastSeq - fs.state.FirstSeq + 1) - fs.state.Msgs)
|
||||
}
|
||||
if numSubjects > numThreshold || numDeleted > numThreshold {
|
||||
if numSubjects > highCardinalityThreshold || numDeleted > highCardinalityThreshold {
|
||||
fs.mu.RUnlock()
|
||||
return errStateTooBig
|
||||
}
|
||||
@@ -11860,11 +11896,9 @@ func (fs *fileStore) _writeFullState(force bool) error {
|
||||
}
|
||||
|
||||
// Update dirty if successful.
|
||||
if err == nil {
|
||||
fs.mu.Lock()
|
||||
fs.dirty -= priorDirty
|
||||
fs.mu.Unlock()
|
||||
}
|
||||
fs.mu.Lock()
|
||||
fs.dirty -= priorDirty
|
||||
fs.mu.Unlock()
|
||||
|
||||
// Attempt to write both files, an error in one should not prevent the other from being written.
|
||||
ttlErr := fs.writeTTLState()
|
||||
@@ -12386,16 +12420,33 @@ func (fs *fileStore) SyncDeleted(dbs DeleteBlocks) error {
|
||||
mdbs := fs.deleteBlocks()
|
||||
fs.readUnlockAllMsgBlocks()
|
||||
|
||||
for _, db := range needsCheck {
|
||||
if dr, ok := db.(*DeleteRange); ok {
|
||||
first, last, _ := dr.State()
|
||||
fs.removeMsgsInRange(first, last, true)
|
||||
for _, db := range dbs {
|
||||
first, last, _ := db.State()
|
||||
if first > lseq {
|
||||
break
|
||||
}
|
||||
|
||||
var prune bool
|
||||
if prune, mdbs = pruneDeleteBlock(db, mdbs); prune {
|
||||
continue
|
||||
}
|
||||
|
||||
var err error
|
||||
if _, ok := db.(*DeleteRange); ok {
|
||||
err = fs.removeMsgsInRange(first, last, true)
|
||||
} else {
|
||||
db.Range(func(dseq uint64) bool {
|
||||
fs.removeMsg(dseq, false, true, false)
|
||||
return true
|
||||
_, err = fs.removeMsg(dseq, false, true, false)
|
||||
// Can continue safely if the message doesn't exist.
|
||||
if err == ErrStoreEOF || err == ErrStoreMsgNotFound {
|
||||
err = nil
|
||||
}
|
||||
return err == nil
|
||||
})
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -12738,7 +12789,7 @@ func (o *consumerFileStore) SetStarting(sseq uint64) error {
|
||||
o.mu.Lock()
|
||||
o.state.Delivered.Stream = sseq
|
||||
o.state.AckFloor.Stream = sseq
|
||||
buf, err := o.encodeState()
|
||||
buf := encodeConsumerState(&o.state)
|
||||
o.mu.Unlock()
|
||||
return o.writeState(buf)
|
||||
}
|
||||
|
||||
@@ -31,6 +31,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/minio/highwayhash"
|
||||
"github.com/hanzoai/pubsub/server/gsl"
|
||||
"github.com/hanzoai/pubsub/server/sysmem"
|
||||
"github.com/hanzoai/pubsub/server/tpm"
|
||||
"github.com/nats-io/nkeys"
|
||||
|
||||
@@ -775,6 +775,7 @@ type JSApiConsumerResetRequest struct {
|
||||
Seq uint64 `json:"seq,omitempty"`
|
||||
}
|
||||
|
||||
// JSApiConsumerResetResponse is a superset of JSApiConsumerCreateResponse, but including an explicit ResetSeq.
|
||||
type JSApiConsumerResetResponse struct {
|
||||
ApiResponse
|
||||
*ConsumerInfo
|
||||
@@ -4313,6 +4314,8 @@ func (s *Server) streamSnapshot(acc *Account, mset *stream, sr *SnapshotResult,
|
||||
|
||||
var hdr []byte
|
||||
chunk := make([]byte, chunkSize)
|
||||
ackTimer := time.NewTimer(snapshotAckTimeout)
|
||||
defer stopAndClearTimer(&ackTimer)
|
||||
for index := 1; ; index++ {
|
||||
select {
|
||||
case <-slots:
|
||||
@@ -4325,7 +4328,7 @@ func (s *Server) streamSnapshot(acc *Account, mset *stream, sr *SnapshotResult,
|
||||
// The snapshotting goroutine has failed for some reason.
|
||||
hdr = []byte(fmt.Sprintf("NATS/1.0 500 %s\r\n\r\n", err))
|
||||
goto done
|
||||
case <-time.After(snapshotAckTimeout):
|
||||
case <-ackTimer.C:
|
||||
// It's taking a very long time for the receiver to send us acks,
|
||||
// they have probably stalled or there is high loss on the link.
|
||||
hdr = []byte("NATS/1.0 408 No Flow Response\r\n\r\n")
|
||||
@@ -4344,6 +4347,7 @@ func (s *Server) streamSnapshot(acc *Account, mset *stream, sr *SnapshotResult,
|
||||
hdr = []byte("NATS/1.0 204\r\n\r\n")
|
||||
}
|
||||
mset.outq.send(newJSPubMsg(reply, _EMPTY_, ackReply, nil, chunk, nil, 0))
|
||||
ackTimer.Reset(snapshotAckTimeout)
|
||||
}
|
||||
|
||||
done:
|
||||
|
||||
+150
-96
@@ -53,6 +53,10 @@ type jetStreamCluster struct {
|
||||
// a response but they need to be same group, peers etc. and sync subjects.
|
||||
inflightStreams map[string]map[string]*inflightStreamInfo
|
||||
inflightConsumers map[string]map[string]map[string]*inflightConsumerInfo
|
||||
// Tracks raft groups currently being started by createRaftGroup, so that
|
||||
// concurrent callers for the same group can wait without holding js.mu
|
||||
// across the disk I/O performed during startup.
|
||||
creatingRaftGroups map[string]chan struct{}
|
||||
// Holds a map of a peer ID to the reply subject, to only respond after gaining
|
||||
// quorum on the peer-remove action.
|
||||
peerRemoveReply map[string]peerRemoveInfo
|
||||
@@ -807,6 +811,14 @@ func (js *jetStream) isConsumerHealthy(mset *stream, consumer string, ca *consum
|
||||
js.mu.RUnlock()
|
||||
return errors.New("consumer assignment or group missing")
|
||||
}
|
||||
// Surface any persisted assignment-level error (e.g. failed create on this
|
||||
// peer) so the health check reflects the broken state instead of falling
|
||||
// through to runtime-only checks.
|
||||
if ca.err != nil {
|
||||
err := ca.err
|
||||
js.mu.RUnlock()
|
||||
return fmt.Errorf("consumer assignment error: %w", err)
|
||||
}
|
||||
created := ca.Created
|
||||
node := ca.Group.node
|
||||
js.mu.RUnlock()
|
||||
@@ -1548,7 +1560,7 @@ func (js *jetStream) getOrphans() (streams []*stream, consumers []*consumer) {
|
||||
streams = append(streams, mset)
|
||||
} else {
|
||||
// This one is good, check consumers now.
|
||||
for _, o := range mset.getConsumers() {
|
||||
for _, o := range mset.getPublicConsumers() {
|
||||
if sa.consumers[o.String()] == nil {
|
||||
consumers = append(consumers, o)
|
||||
}
|
||||
@@ -2166,7 +2178,7 @@ func (js *jetStream) decodeMetaSnapshot(buf []byte) (map[string]map[string]*stre
|
||||
if wca.Stream == _EMPTY_ {
|
||||
wca.Stream = sa.Config.Name // Rehydrate from the stream name.
|
||||
}
|
||||
ca := &consumerAssignment{Client: wca.Client, Created: wca.Created, Name: wca.Name, Stream: wca.Stream, ConfigJSON: wca.ConfigJSON, Group: wca.Group, State: wca.State}
|
||||
ca := &consumerAssignment{Client: wca.Client, Created: wca.Created, Name: wca.Name, Stream: wca.Stream, ConfigJSON: wca.ConfigJSON, Group: wca.Group}
|
||||
if err := decodeConsumerAssignmentConfig(ca); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -2206,7 +2218,6 @@ func (js *jetStream) encodeMetaSnapshot(streams map[string]map[string]*streamAss
|
||||
Stream: ca.Stream,
|
||||
ConfigJSON: ca.ConfigJSON,
|
||||
Group: ca.Group,
|
||||
State: ca.State,
|
||||
}
|
||||
wsa.Consumers = append(wsa.Consumers, &wca)
|
||||
nca++
|
||||
@@ -2328,6 +2339,10 @@ func (js *jetStream) collectStreamAndConsumerChanges(c RaftNodeCheckpoint, strea
|
||||
as = make(map[string]*streamAssignment)
|
||||
streams[sa.Client.serviceAccount()] = as
|
||||
}
|
||||
// Preserve consumers from the previous assignment.
|
||||
if osa := as[sa.Config.Name]; osa != nil {
|
||||
sa.consumers = osa.consumers
|
||||
}
|
||||
as[sa.Config.Name] = sa
|
||||
}
|
||||
for _, cas := range ru.updateConsumers {
|
||||
@@ -2595,7 +2610,9 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
|
||||
}
|
||||
|
||||
if e.Type == EntrySnapshot {
|
||||
js.applyMetaSnapshot(e.Data, ru, isRecovering)
|
||||
if err := js.applyMetaSnapshot(e.Data, ru, isRecovering); err != nil {
|
||||
return isRecovering, didSnap, err
|
||||
}
|
||||
didSnap = true
|
||||
} else if e.Type == EntryRemovePeer {
|
||||
if !js.isMetaRecovering() {
|
||||
@@ -3117,11 +3134,16 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
|
||||
// fully recovered from disk.
|
||||
isRecovering := true
|
||||
|
||||
var failedSnapshots int
|
||||
var (
|
||||
snapMu sync.Mutex
|
||||
snapshotting bool
|
||||
fallbackSnapshot bool
|
||||
failedSnapshots int
|
||||
)
|
||||
|
||||
doSnapshot := func(force bool) {
|
||||
// Suppress during recovery.
|
||||
// If snapshots have failed, and we're not forced to, we'll wait for the timer since it'll now be forced.
|
||||
if mset == nil || isRecovering || isRestore || (!force && failedSnapshots > 0) {
|
||||
if mset == nil || isRecovering || isRestore {
|
||||
return
|
||||
}
|
||||
snapMu.Lock()
|
||||
@@ -3155,28 +3177,70 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
|
||||
}
|
||||
|
||||
// Make sure all pending data is flushed before allowing snapshots.
|
||||
mset.flushAllPending()
|
||||
if err := mset.flushAllPending(); err != nil {
|
||||
// If the pending data couldn't be flushed, we have no safe way to continue.
|
||||
s.Errorf("Failed to flush pending data for '%s > %s' [%s]: %v", accName, mset.name(), n.Group(), err)
|
||||
assert.Unreachable("Stream snapshot flush failed", map[string]any{
|
||||
"account": accName,
|
||||
"stream": mset.name(),
|
||||
"group": n.Group(),
|
||||
"err": err,
|
||||
})
|
||||
c.Abort()
|
||||
mset.setWriteErr(err)
|
||||
n.Stop()
|
||||
return
|
||||
}
|
||||
|
||||
// If we had a significant number of failed snapshots, start relaxing Raft-layer checks
|
||||
// to force it through. We might have been catching up a peer for a long period, and this
|
||||
// protects our log size from growing indefinitely.
|
||||
forceSnapshot := failedSnapshots > 4
|
||||
if err := n.InstallSnapshot(mset.stateSnapshot(), forceSnapshot); err == nil {
|
||||
lastState = curState
|
||||
// If there was a failed snapshot before, we reduced the timer's interval.
|
||||
// Reset it back to the original interval now.
|
||||
if failedSnapshots > 0 {
|
||||
t.Reset(compactInterval + rci)
|
||||
snap := mset.stateSnapshot()
|
||||
|
||||
handleInstallResult := func(err error) {
|
||||
snapshotting = false
|
||||
if err == nil {
|
||||
// If there was a failed snapshot before, we reduced the timer's interval.
|
||||
// Reset it back to the original interval now.
|
||||
if failedSnapshots > 0 {
|
||||
t.Reset(compactInterval + rci)
|
||||
}
|
||||
failedSnapshots = 0
|
||||
fallbackSnapshot = false
|
||||
} else {
|
||||
c.Abort()
|
||||
|
||||
if err == errNoSnapAvailable || err == errNodeClosed || err == errCatchupsRunning || err == errSnapAborted {
|
||||
return
|
||||
}
|
||||
|
||||
s.RateLimitWarnf("Failed to install snapshot for '%s > %s' [%s]: %v, will fall back to blocking snapshot",
|
||||
mset.acc.Name, mset.name(), n.Group(), err)
|
||||
fallbackSnapshot = true
|
||||
// If this is the first failure, reduce the interval of the snapshot timer.
|
||||
// This ensures we're not waiting too long for snapshotting to eventually become forced.
|
||||
if failedSnapshots == 0 {
|
||||
t.Reset(compactMinInterval)
|
||||
}
|
||||
failedSnapshots++
|
||||
}
|
||||
failedSnapshots = 0
|
||||
} else if err != errNoSnapAvailable && err != errNodeClosed && err != errCatchupsRunning {
|
||||
s.RateLimitWarnf("Failed to install snapshot for '%s > %s' [%s]: %v", mset.acc.Name, mset.name(), n.Group(), err)
|
||||
// If this is the first failure, reduce the interval of the snapshot timer.
|
||||
// This ensures we're not waiting too long for snapshotting to eventually become forced.
|
||||
if failedSnapshots == 0 {
|
||||
t.Reset(compactMinInterval)
|
||||
}
|
||||
|
||||
snapshotting = true
|
||||
if fallbackSnapshot {
|
||||
_, err = c.InstallSnapshot(snap)
|
||||
handleInstallResult(err)
|
||||
} else {
|
||||
started := s.startGoRoutine(func() {
|
||||
defer s.grWG.Done()
|
||||
|
||||
_, err := c.InstallSnapshot(snap)
|
||||
|
||||
snapMu.Lock()
|
||||
defer snapMu.Unlock()
|
||||
handleInstallResult(err)
|
||||
})
|
||||
if !started {
|
||||
snapshotting = false
|
||||
c.Abort()
|
||||
}
|
||||
failedSnapshots++
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3253,6 +3317,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
|
||||
select {
|
||||
case <-s.quitCh:
|
||||
// Server shutting down, but we might receive this before qch, so try to snapshot.
|
||||
snapMu.Lock()
|
||||
fallbackSnapshot = true
|
||||
snapMu.Unlock()
|
||||
doSnapshot(false)
|
||||
return
|
||||
case <-mqch:
|
||||
@@ -3260,6 +3327,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
|
||||
// Don't snapshot if not shutting down, monitor goroutine could be going away
|
||||
// on a scale down or a remove for example.
|
||||
if s.isShuttingDown() {
|
||||
snapMu.Lock()
|
||||
fallbackSnapshot = true
|
||||
snapMu.Unlock()
|
||||
doSnapshot(false)
|
||||
}
|
||||
return
|
||||
@@ -3459,7 +3529,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
|
||||
|
||||
case <-t.C:
|
||||
// Start forcing snapshots if they failed previously.
|
||||
snapMu.Lock()
|
||||
forceIfFailed := failedSnapshots > 0
|
||||
snapMu.Unlock()
|
||||
doSnapshot(forceIfFailed)
|
||||
|
||||
case <-uch:
|
||||
@@ -5038,8 +5110,7 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
|
||||
s.Warnf("JetStream cluster detected stream remapping for '%s > %s' from %q to %q",
|
||||
acc, cfg.Name, osa.Group.Name, sa.Group.Name)
|
||||
mset.removeNode()
|
||||
mset.signalMonitorQuit()
|
||||
mset.monitorWg.Wait()
|
||||
mset.stopMonitoring()
|
||||
alreadyRunning, needsNode = false, true
|
||||
// Make sure to clear from original.
|
||||
js.mu.Lock()
|
||||
@@ -5080,8 +5151,7 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
|
||||
} else if numReplicas == 1 && alreadyRunning {
|
||||
// We downgraded to R1. Make sure we cleanup the raft node and the stream monitor.
|
||||
mset.removeNode()
|
||||
mset.signalMonitorQuit()
|
||||
mset.monitorWg.Wait()
|
||||
mset.stopMonitoring()
|
||||
// In case we need to shutdown the cluster specific subs, etc.
|
||||
mset.mu.Lock()
|
||||
// Stop responding to sync requests.
|
||||
@@ -6267,6 +6337,9 @@ func (js *jetStream) consumerAssignmentsOrInflightSeq(account, stream string) it
|
||||
}
|
||||
}
|
||||
sa := js.streamAssignment(account, stream)
|
||||
if sa == nil {
|
||||
return
|
||||
}
|
||||
for _, ca := range sa.consumers {
|
||||
// Skip if we already iterated over it as inflight.
|
||||
if _, ok := inflight[ca.Name]; ok {
|
||||
@@ -6416,33 +6489,16 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
|
||||
}
|
||||
|
||||
if snap, err := o.store.EncodedState(); err == nil {
|
||||
hash := highwayhash.Sum(snap, key)
|
||||
// If the state hasn't changed but the log has gone way over
|
||||
// the compaction size then we will want to compact anyway.
|
||||
// This can happen for example when a pull consumer fetches a
|
||||
// lot on an idle stream, log entries get distributed but the
|
||||
// state never changes, therefore the log never gets compacted.
|
||||
if !bytes.Equal(hash[:], lastSnap) || ne >= compactNumMin || nb >= compactSizeMin {
|
||||
// If we had a significant number of failed snapshots, start relaxing Raft-layer checks
|
||||
// to force it through. We might have been catching up a peer for a long period, and this
|
||||
// protects our log size from growing indefinitely.
|
||||
forceSnapshot := failedSnapshots > 4
|
||||
if err := n.InstallSnapshot(snap, forceSnapshot); err == nil {
|
||||
lastSnap, lastSnapTime = hash[:], time.Now()
|
||||
// If there was a failed snapshot before, we reduced the timer's interval.
|
||||
// Reset it back to the original interval now.
|
||||
if failedSnapshots > 0 {
|
||||
t.Reset(compactInterval + rci)
|
||||
}
|
||||
failedSnapshots = 0
|
||||
} else if err != errNoSnapAvailable && err != errNodeClosed && err != errCatchupsRunning {
|
||||
s.RateLimitWarnf("Failed to install snapshot for '%s > %s > %s' [%s]: %v", o.acc.Name, ca.Stream, ca.Name, n.Group(), err)
|
||||
// If this is the first failure, reduce the interval of the snapshot timer.
|
||||
// This ensures we're not waiting too long for snapshotting to eventually become forced.
|
||||
if failedSnapshots == 0 {
|
||||
t.Reset(compactMinInterval)
|
||||
}
|
||||
failedSnapshots++
|
||||
// If we had a significant number of failed snapshots, start relaxing Raft-layer checks
|
||||
// to force it through. We might have been catching up a peer for a long period, and this
|
||||
// protects our log size from growing indefinitely.
|
||||
forceSnapshot := failedSnapshots > 4
|
||||
if err := n.InstallSnapshot(snap, forceSnapshot); err == nil {
|
||||
lastSnapTime = time.Now()
|
||||
// If there was a failed snapshot before, we reduced the timer's interval.
|
||||
// Reset it back to the original interval now.
|
||||
if failedSnapshots > 0 {
|
||||
t.Reset(compactInterval + rci)
|
||||
}
|
||||
failedSnapshots = 0
|
||||
} else if err != errNoSnapAvailable && err != errNodeClosed && err != errCatchupsRunning {
|
||||
@@ -6785,8 +6841,19 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
|
||||
o.streamNumPending()
|
||||
o.signalNewMessages()
|
||||
s, a := o.srv, o.acc
|
||||
o.mu.Unlock()
|
||||
if reply != _EMPTY_ {
|
||||
if reply == _EMPTY_ {
|
||||
o.mu.Unlock()
|
||||
} else if internal, ok := o.rsm[reply]; !ok {
|
||||
o.mu.Unlock()
|
||||
} else {
|
||||
delete(o.rsm, reply)
|
||||
o.mu.Unlock()
|
||||
|
||||
// Check if the reset request needs to be answered on the system account.
|
||||
// This will happen for replicated sourcing consumers that get reset as part of a create/update.
|
||||
if internal {
|
||||
a = nil
|
||||
}
|
||||
var resp = JSApiConsumerResetResponse{ApiResponse: ApiResponse{Type: JSApiConsumerResetResponseType}}
|
||||
resp.ConsumerInfo = setDynamicConsumerInfoMetadata(o.info())
|
||||
resp.ResetSeq = sseq
|
||||
@@ -7179,8 +7246,9 @@ func (js *jetStream) processStreamAssignmentResults(sub *subscription, c *client
|
||||
}
|
||||
s.Warnf("Stream assignment for '%s > %s' rejected by assigned member: %v", sa.Client.serviceAccount(), sa.Config.Name, apiErr)
|
||||
sa.err = NewJSClusterNotAssignedError()
|
||||
cc.meta.Propose(encodeDeleteStreamAssignment(sa))
|
||||
cc.trackInflightStreamProposal(result.Account, sa, true)
|
||||
if err := cc.meta.Propose(encodeDeleteStreamAssignment(sa)); err == nil {
|
||||
cc.trackInflightStreamProposal(result.Account, sa, true)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -8278,26 +8346,6 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
|
||||
if len(rg.Peers) == 1 {
|
||||
rg.Name = groupNameForStream(rg.Peers, rg.Storage)
|
||||
}
|
||||
for _, peer := range rg.Peers {
|
||||
if len(selected) == newCfg.Replicas {
|
||||
break
|
||||
}
|
||||
if peer == curLeader {
|
||||
continue
|
||||
}
|
||||
if si, ok := s.nodeToInfo.Load(peer); ok && si != nil {
|
||||
if si.(nodeInfo).offline {
|
||||
continue
|
||||
}
|
||||
selected = append(selected, peer)
|
||||
}
|
||||
}
|
||||
rg.Peers = selected
|
||||
// Single nodes are not recorded by the NRG layer so we can rename.
|
||||
// MUST do this, otherwise a scaleup afterward could potentially lead to inconsistencies.
|
||||
if len(rg.Peers) == 1 {
|
||||
rg.Name = groupNameForStream(rg.Peers, rg.Storage)
|
||||
}
|
||||
}
|
||||
|
||||
// Need to remap any consumers.
|
||||
@@ -8450,7 +8498,11 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
|
||||
rg.Preferred = _EMPTY_
|
||||
}
|
||||
|
||||
sa := &streamAssignment{Group: rg, Sync: osa.Sync, Created: osa.Created, Config: newCfg, Subject: subject, Reply: reply, Client: ci}
|
||||
syncSubject := osa.Sync
|
||||
if syncSubject == _EMPTY_ {
|
||||
syncSubject = syncSubjForStream()
|
||||
}
|
||||
sa := &streamAssignment{Group: rg, Sync: syncSubject, Created: osa.Created, Config: newCfg, Subject: subject, Reply: reply, Client: ci}
|
||||
if err := meta.Propose(encodeUpdateStreamAssignment(sa)); err != nil {
|
||||
return
|
||||
}
|
||||
@@ -9280,13 +9332,13 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
|
||||
// we're likely updating an existing consumer, so don't count it. Otherwise
|
||||
// we will incorrectly return NewJSMaximumConsumersLimitError for an update.
|
||||
if oname == _EMPTY_ || js.consumerAssignmentOrInflight(acc.Name, stream, oname) == nil {
|
||||
// Don't count DIRECTS.
|
||||
// Don't count direct/sourcing consumers.
|
||||
total := 0
|
||||
for ca := range js.consumerAssignmentsOrInflightSeq(acc.Name, stream) {
|
||||
if ca.unsupported != nil {
|
||||
continue
|
||||
}
|
||||
if ca.Config != nil && !ca.Config.Direct {
|
||||
if ca.Config != nil && !ca.Config.Direct && !ca.Config.Sourcing {
|
||||
total++
|
||||
}
|
||||
}
|
||||
@@ -9436,10 +9488,16 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
|
||||
s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
|
||||
return
|
||||
}
|
||||
subjects := gatherSubjectFilters(cfg.FilterSubject, cfg.FilterSubjects)
|
||||
for oca := range js.consumerAssignmentsOrInflightSeq(acc.Name, stream) {
|
||||
if oca.Name == oname {
|
||||
if oca.Name == oname || oca.Config.Direct || oca.Config.Sourcing {
|
||||
continue
|
||||
}
|
||||
if len(subjects) == 0 {
|
||||
resp.Error = NewJSConsumerWQMultipleUnfilteredError()
|
||||
s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
|
||||
return
|
||||
}
|
||||
for _, psubj := range gatherSubjectFilters(oca.Config.FilterSubject, oca.Config.FilterSubjects) {
|
||||
for _, subj := range subjects {
|
||||
if SubjectsCollide(subj, psubj) {
|
||||
@@ -9538,14 +9596,6 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
|
||||
if len(nca.Group.Peers) == 1 {
|
||||
nca.Group.Name = groupNameForConsumer(nca.Group.Peers, nca.Group.Storage)
|
||||
}
|
||||
// scale down by removing peers from the end
|
||||
newPeerSet = newPeerSet[len(newPeerSet)-rAfter:]
|
||||
nca.Group.Peers = newPeerSet
|
||||
// Single nodes are not recorded by the NRG layer so we can rename.
|
||||
// MUST do this, otherwise a scaleup afterward could potentially lead to inconsistencies.
|
||||
if len(nca.Group.Peers) == 1 {
|
||||
nca.Group.Name = groupNameForConsumer(nca.Group.Peers, nca.Group.Storage)
|
||||
}
|
||||
}
|
||||
|
||||
// Update config and client info on copy of existing.
|
||||
@@ -10440,9 +10490,13 @@ RETRY:
|
||||
if lseq >= snap.LastSeq {
|
||||
// We MUST ensure all data is flushed up to this point, if the store hadn't already.
|
||||
// Because the snapshot needs to represent what has been persisted.
|
||||
mset.flushAllPending()
|
||||
s.Noticef("Catchup for stream '%s > %s' complete (took %v)", mset.account(), mset.name(), time.Since(start).Round(time.Millisecond))
|
||||
return nil
|
||||
err = mset.flushAllPending()
|
||||
if err == nil {
|
||||
s.Noticef("Catchup for stream '%s > %s' complete (took %v)", mset.account(), mset.name(), time.Since(start).Round(time.Millisecond))
|
||||
} else {
|
||||
s.Noticef("Catchup for stream '%s > %s' errored: %v (took %v)", mset.account(), mset.name(), err, time.Since(start).Round(time.Millisecond))
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// Make sure we do not spin and make things worse.
|
||||
|
||||
@@ -353,6 +353,9 @@ const (
|
||||
// JSMessageSchedulesRollupInvalidErr message schedules invalid rollup
|
||||
JSMessageSchedulesRollupInvalidErr ErrorIdentifier = 10192
|
||||
|
||||
// JSMessageSchedulesSchedulerInvalidErr message schedules invalid scheduler
|
||||
JSMessageSchedulesSchedulerInvalidErr ErrorIdentifier = 10212
|
||||
|
||||
// JSMessageSchedulesSourceInvalidErr message schedules source is invalid
|
||||
JSMessageSchedulesSourceInvalidErr ErrorIdentifier = 10203
|
||||
|
||||
@@ -787,6 +790,7 @@ var (
|
||||
JSMessageSchedulesDisabledErr: {Code: 400, ErrCode: 10188, Description: "message schedules is disabled"},
|
||||
JSMessageSchedulesPatternInvalidErr: {Code: 400, ErrCode: 10189, Description: "message schedules pattern is invalid"},
|
||||
JSMessageSchedulesRollupInvalidErr: {Code: 400, ErrCode: 10192, Description: "message schedules invalid rollup"},
|
||||
JSMessageSchedulesSchedulerInvalidErr: {Code: 400, ErrCode: 10212, Description: "message schedules invalid scheduler"},
|
||||
JSMessageSchedulesSourceInvalidErr: {Code: 400, ErrCode: 10203, Description: "message schedules source is invalid"},
|
||||
JSMessageSchedulesTTLInvalidErr: {Code: 400, ErrCode: 10191, Description: "message schedules invalid per-message TTL"},
|
||||
JSMessageSchedulesTargetInvalidErr: {Code: 400, ErrCode: 10190, Description: "message schedules target is invalid"},
|
||||
@@ -2173,6 +2177,16 @@ func NewJSMessageSchedulesRollupInvalidError(opts ...ErrorOption) *ApiError {
|
||||
return ApiErrors[JSMessageSchedulesRollupInvalidErr]
|
||||
}
|
||||
|
||||
// NewJSMessageSchedulesSchedulerInvalidError creates a new JSMessageSchedulesSchedulerInvalidErr error: "message schedules invalid scheduler"
|
||||
func NewJSMessageSchedulesSchedulerInvalidError(opts ...ErrorOption) *ApiError {
|
||||
eopts := parseOpts(opts)
|
||||
if ae, ok := eopts.err.(*ApiError); ok {
|
||||
return ae
|
||||
}
|
||||
|
||||
return ApiErrors[JSMessageSchedulesSchedulerInvalidErr]
|
||||
}
|
||||
|
||||
// NewJSMessageSchedulesSourceInvalidError creates a new JSMessageSchedulesSourceInvalidErr error: "message schedules source is invalid"
|
||||
func NewJSMessageSchedulesSourceInvalidError(opts ...ErrorOption) *ApiError {
|
||||
eopts := parseOpts(opts)
|
||||
|
||||
@@ -17,7 +17,7 @@ import "strconv"
|
||||
|
||||
const (
|
||||
// JSApiLevel is the maximum supported JetStream API level for this server.
|
||||
JSApiLevel int = 3
|
||||
JSApiLevel int = 4
|
||||
|
||||
JSRequiredLevelMetadataKey = "_nats.req.level"
|
||||
JSServerVersionMetadataKey = "_nats.ver"
|
||||
|
||||
+35
-15
@@ -103,17 +103,6 @@ type RaftNodeCheckpoint interface {
|
||||
InstallSnapshot(data []byte) (uint64, error)
|
||||
}
|
||||
|
||||
// RaftNodeCheckpoint is used as an alternative to a direct InstallSnapshot.
|
||||
// A checkpoint is created from CreateSnapshotCheckpoint and allows installing snapshots asynchronously,
|
||||
// as well as loading the last snapshot or entries between the last snapshot and the one we're about to create.
|
||||
// Abort can be called to cancel the snapshot installation at any time, or InstallSnapshot to install it.
|
||||
type RaftNodeCheckpoint interface {
|
||||
LoadLastSnapshot() (snap []byte, err error)
|
||||
AppendEntriesSeq() iter.Seq2[*appendEntry, error]
|
||||
Abort()
|
||||
InstallSnapshot(data []byte) (uint64, error)
|
||||
}
|
||||
|
||||
type WAL interface {
|
||||
Type() StorageType
|
||||
StoreMsg(subj string, hdr, msg []byte, ttl int64) (uint64, int64, error)
|
||||
@@ -260,6 +249,9 @@ type raft struct {
|
||||
scaleUp bool // The node is part of a scale up, puts us in observer mode until the log contains data.
|
||||
deleted bool // If the node was deleted.
|
||||
snapshotting bool // Snapshot is in progress.
|
||||
quorumPaused bool // Pause replication and quorum participation to prevent log growth during slow applies.
|
||||
|
||||
overrunCount uint64 // Counter of how many times we were overrun, either as follower or as leader.
|
||||
}
|
||||
|
||||
type proposedEntry struct {
|
||||
@@ -1557,12 +1549,20 @@ func (c *checkpoint) InstallSnapshot(data []byte) (uint64, error) {
|
||||
n.Unlock()
|
||||
err := writeFileWithSync(c.snapFile, encoded, defaultFilePerms)
|
||||
n.Lock()
|
||||
// On either failure path, drop the file we just wrote so it doesn't get
|
||||
// picked up by setupLastSnapshot on restart. Skip the remove if it's the
|
||||
// snapshot already adopted into n.snapfile for this term/applied.
|
||||
if err != nil {
|
||||
if c.snapFile != n.snapfile {
|
||||
os.Remove(c.snapFile)
|
||||
}
|
||||
// We could set write err here, but if this is a temporary situation, too many open files etc.
|
||||
// we want to retry and snapshots are not fatal.
|
||||
return 0, err
|
||||
} else if !n.snapshotting {
|
||||
// The checkpoint can be aborted at any time, don't continue if that happened.
|
||||
if c.snapFile != n.snapfile {
|
||||
os.Remove(c.snapFile)
|
||||
}
|
||||
return 0, errSnapAborted
|
||||
}
|
||||
|
||||
@@ -2979,6 +2979,16 @@ func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _
|
||||
n.RUnlock()
|
||||
return
|
||||
}
|
||||
if _, ok := n.peers[string(msg)]; !ok {
|
||||
n.debug("Ignoring forwarded peer removal proposal, peer not found")
|
||||
n.RUnlock()
|
||||
return
|
||||
}
|
||||
if len(n.peers) <= 1 {
|
||||
n.debug("Ignoring forwarded peer removal proposal, remove last node")
|
||||
n.RUnlock()
|
||||
return
|
||||
}
|
||||
prop := n.prop
|
||||
n.RUnlock()
|
||||
|
||||
@@ -3979,6 +3989,19 @@ func (n *raft) truncateWAL(term, index uint64) {
|
||||
// Set after we know we have truncated properly.
|
||||
n.pterm, n.pindex = term, index
|
||||
|
||||
// Invalidate cached entries the WAL no longer has.
|
||||
if index == 0 {
|
||||
if len(n.pae) > 0 {
|
||||
n.pae = make(map[uint64]*appendEntry)
|
||||
}
|
||||
} else {
|
||||
for k := range n.pae {
|
||||
if k > index {
|
||||
delete(n.pae, k)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check if we're truncating an uncommitted membership change.
|
||||
if n.membChangeIndex > 0 && n.membChangeIndex > index {
|
||||
n.membChangeIndex = 0
|
||||
@@ -4333,9 +4356,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
|
||||
|
||||
// Inherit state from appendEntry with the leader's snapshot.
|
||||
hadPreviousSnapshot := n.snapfile != _EMPTY_
|
||||
n.pindex = ae.pindex
|
||||
n.pterm = ae.pterm
|
||||
n.commit = ae.pindex
|
||||
|
||||
snap := &snapshot{
|
||||
lastTerm: ae.pterm,
|
||||
|
||||
+4
-3
@@ -199,11 +199,12 @@ func (ms *MsgScheduling) getScheduledMessages(loadMsg func(seq uint64, smv *Stor
|
||||
ms.remove(seq)
|
||||
return true
|
||||
}
|
||||
rollup := getMessageScheduleRollup(sm.hdr)
|
||||
source := getMessageScheduleSource(sm.hdr)
|
||||
if source != _EMPTY_ {
|
||||
if sm = loadLast(source, &smv); sm == nil {
|
||||
ms.remove(seq)
|
||||
return true
|
||||
// Fall back to the scheduled message's own content if the source has no last message.
|
||||
if srcSm := loadLast(source, &srcSmv); srcSm != nil {
|
||||
sm = srcSm
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+16
-5
@@ -806,11 +806,22 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
|
||||
jsa.mu.Unlock()
|
||||
return nil, NewJSNoLimitsError()
|
||||
}
|
||||
js.mu.RLock()
|
||||
if isClustered {
|
||||
_, reserved = js.tieredStreamAndReservationCount(a.Name, tier, cfg)
|
||||
}
|
||||
if err := js.checkAllLimits(&selected, cfg, reserved, 0); err != nil {
|
||||
|
||||
// Skip if we're recovering.
|
||||
if !recovering {
|
||||
reserved := int64(0)
|
||||
if !isClustered {
|
||||
reserved = jsa.tieredReservation(tier, cfg)
|
||||
}
|
||||
jsa.mu.Unlock()
|
||||
js.mu.RLock()
|
||||
if isClustered {
|
||||
_, reserved = js.tieredStreamAndReservationCount(a.Name, tier, cfg)
|
||||
}
|
||||
if err := js.checkAllLimits(&selected, tier, cfg, reserved, 0); err != nil {
|
||||
js.mu.RUnlock()
|
||||
return nil, err
|
||||
}
|
||||
js.mu.RUnlock()
|
||||
jsa.mu.Lock()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user