diff --git a/docs/svs-v4.md b/docs/svs-v4.md new file mode 100644 index 00000000..c9942724 --- /dev/null +++ b/docs/svs-v4.md @@ -0,0 +1,464 @@ +# State Vector Sync (SVS) v4 Specification + +SVS v4 is a state-vector synchronization protocol for large sync groups. +It introduces a membership hash (`mhash`), two direct state-vector +encodings (`FULL` and `PARTIAL`) carried as distinct TLVs, and a third +publish-only form that references a retrievable full vector. Every Sync +Data carries `mhash` and one of `FullStateVector`, `PartialStateVector`, +or `SvsDataRef`. + +--- + +## 1. Basic Protocol Design + +### 1.1 Small groups + +For most deployments, the complete State Vector fits in one Sync packet. +Nodes exchange **full** State Vectors using steady-state, suppression, merge, +and `OnUpdate` semantics. + +### 1.2 Large groups + +When the encoded State Vector exceeds **`SyncVectorThreshold`** (a +fixed library constant of 1200 bytes), nodes use three dissemination +modes: + +| Mode | Trigger | Wire shape | +|------|---------|------------| +| **Direct FULL** | Encoded FULL fits in threshold | `FullStateVector` (`mhash` + complete `StateVector`) in Sync Data | +| **Direct PARTIAL** | New publication and FULL exceeds threshold | `PartialStateVector` (`mhash` + subset `StateVector`) in Sync Data | +| **Referenced FULL** | Periodic sync (large group), or `mhash` mismatch | Produce full vector Data at `32=sv/`; Sync Data carries only `SvsDataRef` | + +`mhash` is defined in §3.3. + +**Full state recovery** uses publish + pull when: + +1. `mhash` differs from the local membership hash, or +2. Periodic sync runs while the local FULL encoding exceeds + `SyncVectorThreshold`, or +3. A direct `FullStateVector` is outdated per §6.2. + +Retrievable full-vector Data uses the standard NDN segmentation convention +when it exceeds a single packet. + +--- + +## 2. Format and Naming + +### 2.1 Sync Interest + +**Sync Interest Name:** + +``` +//v=4 +``` + +Implementations MAY append additional name components after `v=4`. The +Interest nonce is carried in Interest packet fields, not as a name component. + +- Signed Sync Data is carried in `ApplicationParameters`. +- Interest Lifetime is 1 second. +- Sync Interests are unacknowledged. + +### 2.2 Sync Data (in ApplicationParameters) + +**Sync Data Name** (signing identity for the Sync message): + +``` +//// +``` + +- **`version`:** microsecond timestamp. No hash suffix is used. + +**Sync Data Content:** encoded `SvsData` (§3) — either direct form (FULL +or PARTIAL) or publish-only form. + +### 2.3 Application publication Data + +``` +////seq= +``` + +Application-level naming may vary. Sync vector Data lives in a separate +namespace distinguished by the `32=sv` keyword (§2.4). + +### 2.4 Published full State Vector Data + +Retrievable full State Vector objects use a dedicated sync namespace: + +**Name:** + +``` +////32=sv/ +``` + +**Content:** signed `SvsData` in direct FULL form: `FullStateVector` +containing `mhash` and complete `StateVector`. + +**Publish + pull procedure** (periodic sync, `mhash` recovery, join when +FULL exceeds threshold): + +1. Produce the full-vector Data at + `////32=sv/`. The data is segmented + per the standard NDN convention if it does not fit in a single packet. +2. Send a Sync Interest whose AppParam Sync Data contains publish-only + `SvsData`: `mhash` + `SvsDataRef` pointing at the published name (§3.1). +3. Receivers pull the referenced Data, validate, and merge. + +A Sync message carries either a direct StateVector or a publish-only +reference — not both. + +--- + +## 3. Packet Specification + +### 3.1 `SvsData` + +`SvsData` is a tagged union. The wire carries: + +- `MemberSetHash` (`0xCB`): 32-byte `mhash`, present in all three forms. +- One of the following three top-level TLVs (the choice replaces the + previous `VectorType` discriminator): + - `FullStateVector` (`0xCD`): direct form with a complete State Vector. + - `PartialStateVector` (`0xCE`): direct form with a publication-time + subset. + - `SvsDataRef` (`0x07`): publish-only form pointing to a retrievable + full-vector Data. + +``` +SvsData = SVS-DATA-TYPE TLV-LENGTH + MemberSetHash ; always present + ( FullStateVector + | PartialStateVector + | SvsDataRef ) ; exactly one + +FullStateVector = FULL-STATE-VECTOR-TYPE TLV-LENGTH StateVector +PartialStateVector = PARTIAL-STATE-VECTOR-TYPE TLV-LENGTH StateVector +``` + +| Field | TLV type | Value | +|-------|----------|-------| +| `MemberSetHash` | `0xCB` | 32-byte SHA-256 digest (`mhash`) | +| `FullStateVector` | `0xCD` | Complete `StateVector` (§3.2) | +| `PartialStateVector` | `0xCE` | Subset `StateVector` (§3.2) | +| `SvsDataRef` | `0x07` (Name) | Name of the published full-vector Data. The receiver strips the trailing version component and uses the resulting `32=sv` prefix as the trust anchor for that sender's retrievable full vectors. | + +The receiver MUST reject Sync Data that carries both `FullStateVector` +and `PartialStateVector`, and Sync Data that carries none of the three. +`mhash` MUST be exactly 32 bytes when present. + +### 3.2 `StateVector` + +``` +StateVector = STATE-VECTOR-TYPE TLV-LENGTH + *StateVectorEntry + +StateVectorEntry = STATE-VECTOR-ENTRY-TYPE TLV-LENGTH + Name + *SeqNoEntry + +SeqNoEntry = SEQ-NO-ENTRY-TYPE TLV-LENGTH + BootstrapTime + SeqNo +``` + +| TLV | Type (decimal) | Type (hex) | +|-----|----------------|------------| +| `FULL-STATE-VECTOR-TYPE` | 205 | `0xCD` | +| `PARTIAL-STATE-VECTOR-TYPE` | 206 | `0xCE` | +| `STATE-VECTOR-TYPE` | 201 | `0xC9` | +| `STATE-VECTOR-ENTRY-TYPE` | 202 | `0xCA` | +| `SEQ-NO-ENTRY-TYPE` | 210 | `0xD2` | +| `BOOTSTRAP-TIME-TYPE` | 212 | `0xD4` | +| `SEQ-NO-TYPE` | 214 | `0xD6` | + +**Rules:** + +- Sequence numbers are 1-indexed. +- Bootstrap time is seconds since Unix epoch. +- A missing entry compares as `SeqNo = 0` against a present entry. +- Reject the entire `StateVector` if any received `BootstrapTime` is more + than 86400s in the future. + +### 3.3 `MemberSetHash` (`mhash`) + +`mhash` is a **membership hash**: the SHA-256 digest of the membership set +described below. Membership is independent of sequence numbers, so `mhash` +is unaffected by data publications within the group. + +**Membership** is the set of participants, each identified by: + +``` +(Producer Name, Bootstrap Time) +``` + +**Computation:** + +``` +members = { (Name, BootstrapTime) | node knows this member in the sync group } +sort by NDN canonical order of Name, then by BootstrapTime ascending +mhash = SHA-256( concatenation of canonical TLV bytes of each (Name, BootstrapTime) pair ) +``` + +Recompute `mhash` whenever membership changes (member added, removed, or new +bootstrap time for a name). + +The full State Vector carries membership implicitly: every member's +`StateVectorEntry` is present with its current sequence number. `mhash` +summarizes that membership for quick comparison without having to walk the +full State Vector. + +### 3.4 `FullStateVector` vs. `PartialStateVector` + +The wire TLV itself disambiguates the direct form: + +| TLV | Name | Meaning | +|-----|------|---------| +| `0xCD` | **FULL** | `StateVector` contains the complete advertised state (§4.1 ordering). | +| `0xCE` | **PARTIAL** | `StateVector` contains a subset (§4.2). Used for new publication only when FULL exceeds threshold. | + +Distinct TLVs (instead of a shared field with a discriminator) let a +receiver skip the more expensive subset-evaluation code path when it sees +`FullStateVector`, and let a sender guarantee the receiver knows whether +missing names imply partition (FULL) or merely "not included in this +subset" (PARTIAL). `mhash` alone cannot convey this — two parties with +identical membership but different subscription views may legitimately +disagree on what subset was sent. + +The publish-only form (`SvsDataRef` only, no embedded `StateVector`) +carries neither TLV; the receiver treats it as a signal to fetch the +referenced full vector. + +--- + +## 4. State Vector Encoding + +### 4.1 FULL State Vector + +- Include all known members and their latest sequence numbers per bootstrap. +- Entries ordered in NDN canonical order of `Name`. +- Wire TLV is `FullStateVector` (`0xCD`, §3.4). + +### 4.2 PARTIAL State Vector + +Used on new publication when +`encoded_size(direct FULL SvsData) > SyncVectorThreshold`. + +- Wire TLV is `PartialStateVector` (`0xCE`, §3.4). +- The first entry is the sender's own `StateVectorEntry`; the sender is + always included. +- The remaining entries are the sender's selected peers, ordered in NDN + canonical name order. + +If the sender-only baseline already exceeds `SyncVectorThreshold`, the +sender falls back to publish + pull. The implementation MAY emit an empty +PARTIAL in this case as a signal to the caller; the caller MUST treat an +empty PARTIAL as the publish + pull trigger instead of forwarding it. + +An implementation MAY use the following selection priority: + +| Priority | Include | +|----------|---------| +| 1 | Sender (always) | +| 2 | Repair targets | +| 3 | Propagation targets | +| 4 | Random inactive producers | +| 5 | Others by recency | + +Stop adding entries when the estimated direct `SvsData` size approaches +`SyncVectorThreshold`. + +### 4.3 `SyncVectorThreshold` + +`SyncVectorThreshold` is a fixed library constant (1200 bytes) that bounds +the size of a direct SvsData: + +- When `encoded_size(FULL) ≤ SyncVectorThreshold`, nodes use direct FULL + (`FullStateVector`). +- When `encoded_size(FULL) > SyncVectorThreshold`, nodes switch to PARTIAL + (`PartialStateVector`, on publication) or publish + pull (`SvsDataRef`, + on periodic sync and recovery). + +The wire format is independent of `SyncVectorThreshold`. Every Sync +message carries exactly one of `FullStateVector`, `PartialStateVector`, +or `SvsDataRef`. + +--- + +## 5. State Sync + +Sections 5.1–5.4 describe the steady-state sync loop. Sections 5.5–5.9 +describe the large-group paths. + +### 5.1 Sync Interest timer + +- `PeriodicTimeout` default 30s (±10% jitter). +- `SuppressionPeriod` default 200ms. +- `SuppressionTimeout` exponential decay. + +### 5.2 Send Sync Interest on new publication + +When the node generates a new publication, it immediately emits a Sync +Interest and resets the timer to `PeriodicTimeout`. + +| Trigger | Action | +|---------|--------| +| `encoded_size(direct FULL) ≤ SyncVectorThreshold` | Send direct FULL (`FullStateVector` with `mhash` + `StateVector`) | +| `encoded_size(direct FULL) > SyncVectorThreshold` | Send direct PARTIAL (`PartialStateVector` with `mhash` + subset `StateVector`), or publish + pull if the sender-only baseline itself exceeds the threshold | + +### 5.3 Sync Ack policy + +Sync Interests are unacknowledged. + +### 5.4 Steady state and suppression (direct FULL) + +For incoming Sync Data carrying `FullStateVector`, apply the +steady-state and suppression rules in §5.1–§5.4. + +### 5.5 PARTIAL State Vector processing + +When the wire carries `PartialStateVector`: + +1. Parse `mhash` and `StateVector`. +2. Names omitted from the partial `StateVector` are interpreted as "not + included in this subset" — they do not imply producer removal, outdated + sender, or sequence rollback. +3. For each present entry, merge newer sequence numbers into local state + (§6.1). +4. If `mhash` differs from local `mhash`, perform publish + pull recovery + (§5.6). + +PARTIAL processing is the only receive-side change relative to the +direct-FULL path. + +### 5.6 Full state recovery (publish + pull) + +**Triggers:** + +| # | Trigger | Action | +|---|---------|--------| +| 1 | `mhash` in received `SvsData` ≠ locally computed `mhash` | Publish + pull | +| 2 | Direct `FullStateVector` is outdated per §6.2 | Merge direct if complete; otherwise publish + pull | +| 3 | Periodic sync while local FULL exceeds `SyncVectorThreshold` | Publish + pull (§5.8) | + +Recovery always fetches the complete State Vector from the referenced +`32=sv/` Data. + +**Sender procedure** (on `mhash` mismatch or periodic large-group sync): + +1. Produce full-vector Data at `////32=sv/` + with `FullStateVector` SvsData. +2. Send Sync Interest with publish-only `SvsData` (`SvsDataRef` only). + +**Receiver procedure:** + +1. Identify the sender from the Sync Data signature, or — when the Sync + Data is PARTIAL — from the first entry, which is the sender's own + entry per §4.2. +2. If the Sync Data is direct FULL and complete: merge directly. +3. If the Sync Data is publish-only: read `SvsDataRef`; express Interest for + that name; validate; merge; update local `mhash`. +4. Continue application data fetch via SvsALO (`OnUpdate`) as today. + +> **Implementation note:** A consumer may receive many publish-only Sync +> messages that all cross the `mhash` boundary simultaneously. To bound the +> resulting pull fan-in, implementations commonly debounce per-sender pull +> attempts (e.g., 5 seconds per sender prefix). This is a local +> implementation detail and does not affect protocol correctness — a +> debounced pull is equivalent to a slightly delayed pull. + +Fetched Data is segmented per the standard NDN convention when it does +not fit in a single packet. + +### 5.7 New node join + +1. Joining node **N** multicasts Sync Interest whose direct State Vector + contains only itself: `(Name=N, SeqNo=0)`. The Sync Data's `mhash` is + the SHA-256 of N's membership set, which is the single member `{N}`. +2. Existing members receive the announcement. +3. Suppression limits duplicate responses; typically one member **A** + provides recovery state. +4. If FULL fits in a direct packet: **A** responds with `FullStateVector`. +5. If FULL exceeds `SyncVectorThreshold`: **A** uses publish + pull + (produce at `32=sv/`, then publish-only Sync Data). +6. Normal synchronization proceeds through SvsALO. + +### 5.8 Periodic sync in large groups + +| Local FULL size | Periodic Sync behavior | +|-----------------|------------------------| +| `≤ SyncVectorThreshold` | Direct FULL (`FullStateVector`) | +| `> SyncVectorThreshold` | Publish + pull (produce full-vector Data, then publish-only Sync Data) | + +Periodic sync does not send direct PARTIAL vectors. + +### 5.9 Summary of sync triggers + +| Event | `size ≤ threshold` | `size > threshold` | +|-------|--------------------|--------------------| +| **New publication** | Direct FULL (`FullStateVector`) | Direct PARTIAL (`PartialStateVector`, or publish + pull fallback) | +| **Periodic sync** | Direct FULL (`FullStateVector`) | Publish + pull | +| **`mhash` mismatch** | Publish + pull (if recovery needed) | Publish + pull | + +--- + +## 6. Comparing and Merging State Vectors + +### 6.1 Merge rule + +For each matching `(Name, BootstrapTime)`, retain the maximum `SeqNo`. + +### 6.2 Outdated vector + +State Vector `A` is outdated to `B` if: + +- `A` is missing a name present in `B`, or +- `A` has a strictly smaller `SeqNo` for any entry. + +This rule applies when `A` is a `FullStateVector`. When `A` is a +`PartialStateVector`, `A`'s omitted names are a subset by design (§4.2): +the sender selected a publication-time subset and `A`'s missing entries +do not carry any information about whether `A` is outdated relative to +`B`. + +--- + +## 7. Examples + +### 7.1 Small group + +Three nodes `A`, `B`, `C`. Full State Vector fits. `A` publishes; sends +direct FULL Sync Interest `[A:11, B:15, C:25]`. Peers merge. + +### 7.2 Large group + +Group exceeds `SyncVectorThreshold`. Producer `P` publishes: + +- `P` sends direct PARTIAL `SvsData` carrying `PartialStateVector { mhash, + StateVector=[P:…, A:…, …] }`. +- Receiver merges present entries only. +- If `mhash` differs, `P` (or receiver per policy) triggers publish + pull + (§5.6). + +### 7.3 Large group + +- `A` produces full vector at `/group/A/boot/32=sv/`. +- `A` sends publish-only Sync Data `{ SvsDataRef=/group/A/boot/32=sv/ }`. +- Peers pull and merge. + +### 7.4 New node join + +- `N` sends a State Vector containing only itself (`[N:0]`) with `mhash` + computed over the single-member membership set `{N}`. +- `A` responds with `FullStateVector` or publish + pull. +- `N` merges and synchronizes via SvsALO. + +--- + +## 8. Interoperability + +SVS v4 defines a single wire profile. Deployments upgrade all nodes in a +sync group at the same time. Every Sync Data carries exactly one of +`FullStateVector`, `PartialStateVector`, or `SvsDataRef`. The +implementation never emits a bare `StateVector`-only `SvsData`, +regardless of `SyncVectorThreshold`. \ No newline at end of file diff --git a/dv/dv/advert_sync.go b/dv/dv/advert_sync.go index 5291a244..8fc43360 100644 --- a/dv/dv/advert_sync.go +++ b/dv/dv/advert_sync.go @@ -9,7 +9,7 @@ import ( "github.com/named-data/ndnd/std/log" "github.com/named-data/ndnd/std/ndn" spec "github.com/named-data/ndnd/std/ndn/spec_2022" - spec_svs "github.com/named-data/ndnd/std/ndn/svs/v3" + spec_svs "github.com/named-data/ndnd/std/ndn/svs/v4" "github.com/named-data/ndnd/std/object/storage" "github.com/named-data/ndnd/std/types/optional" "github.com/named-data/ndnd/std/utils" @@ -50,17 +50,18 @@ func (a *advertModule) sendSyncInterest() (err error) { // (AI GENERATED DESCRIPTION): Sends a signed state‑vector Data packet as the payload of a sync Interest to the given `syncName`, expressing the Interest locally without expecting a reply. func (a *advertModule) sendSyncInterestImpl(syncName enc.Name) (err error) { - // State Vector for our group - sv := &spec_svs.SvsData{ - StateVector: &spec_svs.StateVector{ - Entries: []*spec_svs.StateVectorEntry{{ - Name: a.dv.config.RouterName(), - SeqNoEntries: []*spec_svs.SeqNoEntry{{ - BootstrapTime: a.bootTime, - SeqNo: a.seq, - }}, + // DV's advertisement Sync Data carries a single-entry StateVector + // directly. SVS v4's wire (FullStateVector/PartialStateVector/SvsDataRef + // tagged union) is a different protocol layer; DV keeps its own simpler + // shape to remain independent of SVS v4 changes. + sv := &spec_svs.StateVector{ + Entries: []*spec_svs.StateVectorEntry{{ + Name: a.dv.config.RouterName(), + SeqNoEntries: []*spec_svs.SeqNoEntry{{ + BootstrapTime: a.bootTime, + SeqNo: a.seq, }}, - }, + }}, } // Sign the Sync Data @@ -136,14 +137,14 @@ func (a *advertModule) OnSyncInterest(args ndn.InterestHandlerArgs, active bool) // Decode state vector svWire := data.Content() - params, err := spec_svs.ParseSvsData(enc.NewWireView(svWire), false) - if err != nil || params.StateVector == nil { + params, err := spec_svs.ParseStateVector(enc.NewWireView(svWire), false) + if err != nil { log.Warn(a, "Failed to parse StateVec", "err", err) return } // Process the state vector - go a.onStateVector(params.StateVector, args.IncomingFaceId.Unwrap(), active) + go a.onStateVector(params, args.IncomingFaceId.Unwrap(), active) }, }) } diff --git a/e2e/dv_util.py b/e2e/dv_util.py index c3c020f1..86e6eed7 100644 --- a/e2e/dv_util.py +++ b/e2e/dv_util.py @@ -15,7 +15,7 @@ def setup(ndn: Minindn, network=DEFAULT_NETWORK) -> None: info('Starting ndn-dv on nodes\n') AppManager(ndn, ndn.net.hosts, NDNd_DV, network=network) -def converge(nodes: list[Node], deadline=30, network=DEFAULT_NETWORK, use_nfdc=False) -> int: +def converge(nodes: list[Node], deadline=90, network=DEFAULT_NETWORK, use_nfdc=False) -> int: info('Waiting for routing to converge\n') start = time.time() while time.time() - start < deadline: diff --git a/e2e/test_001.py b/e2e/test_001.py index b933ca9d..6123da34 100644 --- a/e2e/test_001.py +++ b/e2e/test_001.py @@ -43,7 +43,7 @@ def scenario(ndn: Minindn, fw=None, network='/minindn'): node.cmd(cmd) info('Waiting for put to complete\n') - time.sleep(30) + time.sleep(90) for node in cat_nodes: put_node = random.choice(put_nodes) diff --git a/std/ndn/svs/v3/definitions.go b/std/ndn/svs/v3/definitions.go deleted file mode 100644 index 3f7777bc..00000000 --- a/std/ndn/svs/v3/definitions.go +++ /dev/null @@ -1,36 +0,0 @@ -//go:generate gondn_tlv_gen -package svs - -import ( - enc "github.com/named-data/ndnd/std/encoding" -) - -type SvsData struct { - //+field:struct:StateVector - StateVector *StateVector `tlv:"0xc9"` -} - -type StateVector struct { - //+field:sequence:*StateVectorEntry:struct:StateVectorEntry - Entries []*StateVectorEntry `tlv:"0xca"` -} - -type StateVectorEntry struct { - //+field:name - Name enc.Name `tlv:"0x07"` - //+field:sequence:*SeqNoEntry:struct:SeqNoEntry - SeqNoEntries []*SeqNoEntry `tlv:"0xd2"` -} - -type SeqNoEntry struct { - //+field:natural - BootstrapTime uint64 `tlv:"0xd4"` - //+field:natural - SeqNo uint64 `tlv:"0xd6"` -} - -// +tlv-model:nocopy -type PassiveState struct { - //+field:sequence:[]byte:binary:[]byte - Data [][]byte `tlv:"0xfa0"` -} diff --git a/std/ndn/svs/v4/accessors.go b/std/ndn/svs/v4/accessors.go new file mode 100644 index 00000000..06cdef63 --- /dev/null +++ b/std/ndn/svs/v4/accessors.go @@ -0,0 +1,54 @@ +package svs + +// VectorKind identifies which direct form an SvsData carries on the wire. +// Publish-only Sync Data carries neither FullStateVector nor +// PartialStateVector; instead it has SvsDataRef. The choice of wire TLV +// replaces the previous VectorType discriminator field. +type VectorKind int + +const ( + // VectorKindNone is the publish-only form: no embedded vector, only a + // retrievable SvsDataRef. + VectorKindNone VectorKind = iota + // VectorKindFull is a complete State Vector (FULL). + VectorKindFull + // VectorKindPartial is a sender-selected subset (PARTIAL). + VectorKindPartial +) + +// GetStateVector returns the embedded StateVector for direct forms (FULL +// or PARTIAL), or nil for the publish-only form. +func (d *SvsData) GetStateVector() *StateVector { + switch { + case d.FullStateVector != nil: + return d.FullStateVector.StateVector + case d.PartialStateVector != nil: + return d.PartialStateVector.StateVector + } + return nil +} + +// Kind reports which direct form (or none) the SvsData carries on the wire. +// An SvsData carrying both FullStateVector and PartialStateVector is +// reported as KindFull (the more specific case wins). +func (d *SvsData) Kind() VectorKind { + switch { + case d.FullStateVector != nil: + return VectorKindFull + case d.PartialStateVector != nil: + return VectorKindPartial + } + return VectorKindNone +} + +// IsPartial reports whether the embedded StateVector is a PARTIAL subset. +// Returns false for the publish-only form (callers that branch on this +// should also check SvsDataRef for the publish-only recovery path). +func (d *SvsData) IsPartial() bool { + return d.PartialStateVector != nil +} + +// IsFull reports whether the embedded StateVector is a FULL State Vector. +func (d *SvsData) IsFull() bool { + return d.FullStateVector != nil +} diff --git a/std/ndn/svs/v4/definitions.go b/std/ndn/svs/v4/definitions.go new file mode 100644 index 00000000..08d12832 --- /dev/null +++ b/std/ndn/svs/v4/definitions.go @@ -0,0 +1,72 @@ +//go:generate gondn_tlv_gen +package svs + +import ( + enc "github.com/named-data/ndnd/std/encoding" +) + +// FullStateVector is the wire form of a complete State Vector carried in a +// Sync message or published at .../32=sv/. The presence of this +// TLV (rather than PartialStateVector) tells the receiver that the embedded +// StateVector represents the sender's full membership view. +type FullStateVector struct { + //+field:struct:StateVector + StateVector *StateVector `tlv:"0xc9"` +} + +// PartialStateVector is the wire form of a publication-time subset State +// Vector. The presence of this TLV tells the receiver that omitted entries +// are a sender-selected subset (not a partition or out-of-date sender). +type PartialStateVector struct { + //+field:struct:StateVector + StateVector *StateVector `tlv:"0xc9"` +} + +// SvsData is a tagged union: the wire carries exactly one of +// FullStateVector, PartialStateVector, or SvsDataRef. The choice of TLV +// type replaces the previous VectorType discriminator. MemberSetHash +// (`mhash`) is present on all three forms and lets receivers detect +// membership mismatches without walking a full StateVector. +type SvsData struct { + //+field:binary:optional + MemberSetHash []byte `tlv:"0xcb"` + //+field:struct:FullStateVector + FullStateVector *FullStateVector `tlv:"0xcd"` + //+field:struct:PartialStateVector + PartialStateVector *PartialStateVector `tlv:"0xce"` + //+field:name + SvsDataRef enc.Name `tlv:"0x07"` +} + +type StateVector struct { + //+field:sequence:*StateVectorEntry:struct:StateVectorEntry + Entries []*StateVectorEntry `tlv:"0xca"` +} + +type StateVectorEntry struct { + //+field:name + Name enc.Name `tlv:"0x07"` + //+field:sequence:*SeqNoEntry:struct:SeqNoEntry + SeqNoEntries []*SeqNoEntry `tlv:"0xd2"` +} + +type SeqNoEntry struct { + //+field:natural + BootstrapTime uint64 `tlv:"0xd4"` + //+field:natural + SeqNo uint64 `tlv:"0xd6"` +} + +// MembershipTuple is one (Name, BootstrapTime) pair used to compute MemberSetHash. +type MembershipTuple struct { + //+field:name + Name enc.Name `tlv:"0x07"` + //+field:natural + BootstrapTime uint64 `tlv:"0xd4"` +} + +// +tlv-model:nocopy +type PassiveState struct { + //+field:sequence:[]byte:binary:[]byte + Data [][]byte `tlv:"0xfa0"` +} diff --git a/std/ndn/svs/v3/zz_generated.go b/std/ndn/svs/v4/zz_generated.go similarity index 62% rename from std/ndn/svs/v3/zz_generated.go rename to std/ndn/svs/v4/zz_generated.go index 0fb2e7fc..230120be 100644 --- a/std/ndn/svs/v3/zz_generated.go +++ b/std/ndn/svs/v4/zz_generated.go @@ -8,17 +8,17 @@ import ( enc "github.com/named-data/ndnd/std/encoding" ) -type SvsDataEncoder struct { +type FullStateVectorEncoder struct { Length uint StateVector_encoder StateVectorEncoder } -type SvsDataParsingContext struct { +type FullStateVectorParsingContext struct { StateVector_context StateVectorParsingContext } -func (encoder *SvsDataEncoder) Init(value *SvsData) { +func (encoder *FullStateVectorEncoder) Init(value *FullStateVector) { if value.StateVector != nil { encoder.StateVector_encoder.Init(value.StateVector) } @@ -33,11 +33,11 @@ func (encoder *SvsDataEncoder) Init(value *SvsData) { } -func (context *SvsDataParsingContext) Init() { +func (context *FullStateVectorParsingContext) Init() { context.StateVector_context.Init() } -func (encoder *SvsDataEncoder) EncodeInto(value *SvsData, buf []byte) { +func (encoder *FullStateVectorEncoder) EncodeInto(value *FullStateVector, buf []byte) { pos := uint(0) @@ -52,7 +52,7 @@ func (encoder *SvsDataEncoder) EncodeInto(value *SvsData, buf []byte) { } } -func (encoder *SvsDataEncoder) Encode(value *SvsData) enc.Wire { +func (encoder *FullStateVectorEncoder) Encode(value *FullStateVector) enc.Wire { wire := make(enc.Wire, 1) wire[0] = make([]byte, encoder.Length) @@ -62,14 +62,14 @@ func (encoder *SvsDataEncoder) Encode(value *SvsData) enc.Wire { return wire } -func (context *SvsDataParsingContext) Parse(reader enc.WireView, ignoreCritical bool) (*SvsData, error) { +func (context *FullStateVectorParsingContext) Parse(reader enc.WireView, ignoreCritical bool) (*FullStateVector, error) { var handled_StateVector bool = false progress := -1 _ = progress - value := &SvsData{} + value := &FullStateVector{} var err error var startPos int for { @@ -126,6 +126,361 @@ func (context *SvsDataParsingContext) Parse(reader enc.WireView, ignoreCritical return value, nil } +func (value *FullStateVector) Encode() enc.Wire { + encoder := FullStateVectorEncoder{} + encoder.Init(value) + return encoder.Encode(value) +} + +func (value *FullStateVector) Bytes() []byte { + return value.Encode().Join() +} + +func ParseFullStateVector(reader enc.WireView, ignoreCritical bool) (*FullStateVector, error) { + context := FullStateVectorParsingContext{} + context.Init() + return context.Parse(reader, ignoreCritical) +} + +type PartialStateVectorEncoder struct { + Length uint + + StateVector_encoder StateVectorEncoder +} + +type PartialStateVectorParsingContext struct { + StateVector_context StateVectorParsingContext +} + +func (encoder *PartialStateVectorEncoder) Init(value *PartialStateVector) { + if value.StateVector != nil { + encoder.StateVector_encoder.Init(value.StateVector) + } + + l := uint(0) + if value.StateVector != nil { + l += 1 + l += uint(enc.TLNum(encoder.StateVector_encoder.Length).EncodingLength()) + l += encoder.StateVector_encoder.Length + } + encoder.Length = l + +} + +func (context *PartialStateVectorParsingContext) Init() { + context.StateVector_context.Init() +} + +func (encoder *PartialStateVectorEncoder) EncodeInto(value *PartialStateVector, buf []byte) { + + pos := uint(0) + + if value.StateVector != nil { + buf[pos] = byte(201) + pos += 1 + pos += uint(enc.TLNum(encoder.StateVector_encoder.Length).EncodeInto(buf[pos:])) + if encoder.StateVector_encoder.Length > 0 { + encoder.StateVector_encoder.EncodeInto(value.StateVector, buf[pos:]) + pos += encoder.StateVector_encoder.Length + } + } +} + +func (encoder *PartialStateVectorEncoder) Encode(value *PartialStateVector) enc.Wire { + + wire := make(enc.Wire, 1) + wire[0] = make([]byte, encoder.Length) + buf := wire[0] + encoder.EncodeInto(value, buf) + + return wire +} + +func (context *PartialStateVectorParsingContext) Parse(reader enc.WireView, ignoreCritical bool) (*PartialStateVector, error) { + + var handled_StateVector bool = false + + progress := -1 + _ = progress + + value := &PartialStateVector{} + var err error + var startPos int + for { + startPos = reader.Pos() + if startPos >= reader.Length() { + break + } + typ := enc.TLNum(0) + l := enc.TLNum(0) + typ, err = reader.ReadTLNum() + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: 0, Err: err} + } + l, err = reader.ReadTLNum() + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: 0, Err: err} + } + + err = nil + if handled := false; true { + switch typ { + case 201: + if true { + handled = true + handled_StateVector = true + value.StateVector, err = context.StateVector_context.Parse(reader.Delegate(int(l)), ignoreCritical) + } + default: + if !ignoreCritical && ((typ <= 31) || ((typ & 1) == 1)) { + return nil, enc.ErrUnrecognizedField{TypeNum: typ} + } + handled = true + err = reader.Skip(int(l)) + } + if err == nil && !handled { + } + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: typ, Err: err} + } + } + } + + startPos = reader.Pos() + err = nil + + if !handled_StateVector && err == nil { + value.StateVector = nil + } + + if err != nil { + return nil, err + } + + return value, nil +} + +func (value *PartialStateVector) Encode() enc.Wire { + encoder := PartialStateVectorEncoder{} + encoder.Init(value) + return encoder.Encode(value) +} + +func (value *PartialStateVector) Bytes() []byte { + return value.Encode().Join() +} + +func ParsePartialStateVector(reader enc.WireView, ignoreCritical bool) (*PartialStateVector, error) { + context := PartialStateVectorParsingContext{} + context.Init() + return context.Parse(reader, ignoreCritical) +} + +type SvsDataEncoder struct { + Length uint + + FullStateVector_encoder FullStateVectorEncoder + PartialStateVector_encoder PartialStateVectorEncoder + SvsDataRef_length uint +} + +type SvsDataParsingContext struct { + FullStateVector_context FullStateVectorParsingContext + PartialStateVector_context PartialStateVectorParsingContext +} + +func (encoder *SvsDataEncoder) Init(value *SvsData) { + + if value.FullStateVector != nil { + encoder.FullStateVector_encoder.Init(value.FullStateVector) + } + if value.PartialStateVector != nil { + encoder.PartialStateVector_encoder.Init(value.PartialStateVector) + } + if value.SvsDataRef != nil { + encoder.SvsDataRef_length = 0 + for _, c := range value.SvsDataRef { + encoder.SvsDataRef_length += uint(c.EncodingLength()) + } + } + + l := uint(0) + if value.MemberSetHash != nil { + l += 1 + l += uint(enc.TLNum(len(value.MemberSetHash)).EncodingLength()) + l += uint(len(value.MemberSetHash)) + } + if value.FullStateVector != nil { + l += 1 + l += uint(enc.TLNum(encoder.FullStateVector_encoder.Length).EncodingLength()) + l += encoder.FullStateVector_encoder.Length + } + if value.PartialStateVector != nil { + l += 1 + l += uint(enc.TLNum(encoder.PartialStateVector_encoder.Length).EncodingLength()) + l += encoder.PartialStateVector_encoder.Length + } + if value.SvsDataRef != nil { + l += 1 + l += uint(enc.TLNum(encoder.SvsDataRef_length).EncodingLength()) + l += encoder.SvsDataRef_length + } + encoder.Length = l + +} + +func (context *SvsDataParsingContext) Init() { + + context.FullStateVector_context.Init() + context.PartialStateVector_context.Init() + +} + +func (encoder *SvsDataEncoder) EncodeInto(value *SvsData, buf []byte) { + + pos := uint(0) + + if value.MemberSetHash != nil { + buf[pos] = byte(203) + pos += 1 + pos += uint(enc.TLNum(len(value.MemberSetHash)).EncodeInto(buf[pos:])) + copy(buf[pos:], value.MemberSetHash) + pos += uint(len(value.MemberSetHash)) + } + if value.FullStateVector != nil { + buf[pos] = byte(205) + pos += 1 + pos += uint(enc.TLNum(encoder.FullStateVector_encoder.Length).EncodeInto(buf[pos:])) + if encoder.FullStateVector_encoder.Length > 0 { + encoder.FullStateVector_encoder.EncodeInto(value.FullStateVector, buf[pos:]) + pos += encoder.FullStateVector_encoder.Length + } + } + if value.PartialStateVector != nil { + buf[pos] = byte(206) + pos += 1 + pos += uint(enc.TLNum(encoder.PartialStateVector_encoder.Length).EncodeInto(buf[pos:])) + if encoder.PartialStateVector_encoder.Length > 0 { + encoder.PartialStateVector_encoder.EncodeInto(value.PartialStateVector, buf[pos:]) + pos += encoder.PartialStateVector_encoder.Length + } + } + if value.SvsDataRef != nil { + buf[pos] = byte(7) + pos += 1 + pos += uint(enc.TLNum(encoder.SvsDataRef_length).EncodeInto(buf[pos:])) + for _, c := range value.SvsDataRef { + pos += uint(c.EncodeInto(buf[pos:])) + } + } +} + +func (encoder *SvsDataEncoder) Encode(value *SvsData) enc.Wire { + + wire := make(enc.Wire, 1) + wire[0] = make([]byte, encoder.Length) + buf := wire[0] + encoder.EncodeInto(value, buf) + + return wire +} + +func (context *SvsDataParsingContext) Parse(reader enc.WireView, ignoreCritical bool) (*SvsData, error) { + + var handled_MemberSetHash bool = false + var handled_FullStateVector bool = false + var handled_PartialStateVector bool = false + var handled_SvsDataRef bool = false + + progress := -1 + _ = progress + + value := &SvsData{} + var err error + var startPos int + for { + startPos = reader.Pos() + if startPos >= reader.Length() { + break + } + typ := enc.TLNum(0) + l := enc.TLNum(0) + typ, err = reader.ReadTLNum() + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: 0, Err: err} + } + l, err = reader.ReadTLNum() + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: 0, Err: err} + } + + err = nil + if handled := false; true { + switch typ { + case 203: + if true { + handled = true + handled_MemberSetHash = true + value.MemberSetHash = make([]byte, l) + _, err = reader.ReadFull(value.MemberSetHash) + } + case 205: + if true { + handled = true + handled_FullStateVector = true + value.FullStateVector, err = context.FullStateVector_context.Parse(reader.Delegate(int(l)), ignoreCritical) + } + case 206: + if true { + handled = true + handled_PartialStateVector = true + value.PartialStateVector, err = context.PartialStateVector_context.Parse(reader.Delegate(int(l)), ignoreCritical) + } + case 7: + if true { + handled = true + handled_SvsDataRef = true + delegate := reader.Delegate(int(l)) + value.SvsDataRef, err = delegate.ReadName() + } + default: + if !ignoreCritical && ((typ <= 31) || ((typ & 1) == 1)) { + return nil, enc.ErrUnrecognizedField{TypeNum: typ} + } + handled = true + err = reader.Skip(int(l)) + } + if err == nil && !handled { + } + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: typ, Err: err} + } + } + } + + startPos = reader.Pos() + err = nil + + if !handled_MemberSetHash && err == nil { + value.MemberSetHash = nil + } + if !handled_FullStateVector && err == nil { + value.FullStateVector = nil + } + if !handled_PartialStateVector && err == nil { + value.PartialStateVector = nil + } + if !handled_SvsDataRef && err == nil { + value.SvsDataRef = nil + } + + if err != nil { + return nil, err + } + + return value, nil +} + func (value *SvsData) Encode() enc.Wire { encoder := SvsDataEncoder{} encoder.Init(value) @@ -742,6 +1097,172 @@ func ParseSeqNoEntry(reader enc.WireView, ignoreCritical bool) (*SeqNoEntry, err return context.Parse(reader, ignoreCritical) } +type MembershipTupleEncoder struct { + Length uint + + Name_length uint +} + +type MembershipTupleParsingContext struct { +} + +func (encoder *MembershipTupleEncoder) Init(value *MembershipTuple) { + if value.Name != nil { + encoder.Name_length = 0 + for _, c := range value.Name { + encoder.Name_length += uint(c.EncodingLength()) + } + } + + l := uint(0) + if value.Name != nil { + l += 1 + l += uint(enc.TLNum(encoder.Name_length).EncodingLength()) + l += encoder.Name_length + } + l += 1 + l += uint(1 + enc.Nat(value.BootstrapTime).EncodingLength()) + encoder.Length = l + +} + +func (context *MembershipTupleParsingContext) Init() { + +} + +func (encoder *MembershipTupleEncoder) EncodeInto(value *MembershipTuple, buf []byte) { + + pos := uint(0) + + if value.Name != nil { + buf[pos] = byte(7) + pos += 1 + pos += uint(enc.TLNum(encoder.Name_length).EncodeInto(buf[pos:])) + for _, c := range value.Name { + pos += uint(c.EncodeInto(buf[pos:])) + } + } + buf[pos] = byte(212) + pos += 1 + + buf[pos] = byte(enc.Nat(value.BootstrapTime).EncodeInto(buf[pos+1:])) + pos += uint(1 + buf[pos]) +} + +func (encoder *MembershipTupleEncoder) Encode(value *MembershipTuple) enc.Wire { + + wire := make(enc.Wire, 1) + wire[0] = make([]byte, encoder.Length) + buf := wire[0] + encoder.EncodeInto(value, buf) + + return wire +} + +func (context *MembershipTupleParsingContext) Parse(reader enc.WireView, ignoreCritical bool) (*MembershipTuple, error) { + + var handled_Name bool = false + var handled_BootstrapTime bool = false + + progress := -1 + _ = progress + + value := &MembershipTuple{} + var err error + var startPos int + for { + startPos = reader.Pos() + if startPos >= reader.Length() { + break + } + typ := enc.TLNum(0) + l := enc.TLNum(0) + typ, err = reader.ReadTLNum() + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: 0, Err: err} + } + l, err = reader.ReadTLNum() + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: 0, Err: err} + } + + err = nil + if handled := false; true { + switch typ { + case 7: + if true { + handled = true + handled_Name = true + delegate := reader.Delegate(int(l)) + value.Name, err = delegate.ReadName() + } + case 212: + if true { + handled = true + handled_BootstrapTime = true + value.BootstrapTime = uint64(0) + { + for i := 0; i < int(l); i++ { + x := byte(0) + x, err = reader.ReadByte() + if err != nil { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + break + } + value.BootstrapTime = uint64(value.BootstrapTime<<8) | uint64(x) + } + } + } + default: + if !ignoreCritical && ((typ <= 31) || ((typ & 1) == 1)) { + return nil, enc.ErrUnrecognizedField{TypeNum: typ} + } + handled = true + err = reader.Skip(int(l)) + } + if err == nil && !handled { + } + if err != nil { + return nil, enc.ErrFailToParse{TypeNum: typ, Err: err} + } + } + } + + startPos = reader.Pos() + err = nil + + if !handled_Name && err == nil { + value.Name = nil + } + if !handled_BootstrapTime && err == nil { + err = enc.ErrSkipRequired{Name: "BootstrapTime", TypeNum: 212} + } + + if err != nil { + return nil, err + } + + return value, nil +} + +func (value *MembershipTuple) Encode() enc.Wire { + encoder := MembershipTupleEncoder{} + encoder.Init(value) + return encoder.Encode(value) +} + +func (value *MembershipTuple) Bytes() []byte { + return value.Encode().Join() +} + +func ParseMembershipTuple(reader enc.WireView, ignoreCritical bool) (*MembershipTuple, error) { + context := MembershipTupleParsingContext{} + context.Init() + return context.Parse(reader, ignoreCritical) +} + type PassiveStateEncoder struct { Length uint diff --git a/std/ndn/svs_ps/definitions.go b/std/ndn/svs_ps/definitions.go index 2e3f879c..900ed9d4 100644 --- a/std/ndn/svs_ps/definitions.go +++ b/std/ndn/svs_ps/definitions.go @@ -3,7 +3,7 @@ package svs_ps import ( enc "github.com/named-data/ndnd/std/encoding" - "github.com/named-data/ndnd/std/ndn/svs/v3" + "github.com/named-data/ndnd/std/ndn/svs/v4" ) type InstanceState struct { diff --git a/std/ndn/svs_ps/zz_generated.go b/std/ndn/svs_ps/zz_generated.go index c16c1abc..2fdaae5a 100644 --- a/std/ndn/svs_ps/zz_generated.go +++ b/std/ndn/svs_ps/zz_generated.go @@ -5,7 +5,7 @@ import ( "io" enc "github.com/named-data/ndnd/std/encoding" - "github.com/named-data/ndnd/std/ndn/svs/v3" + "github.com/named-data/ndnd/std/ndn/svs/v4" ) type InstanceStateEncoder struct { diff --git a/std/object/client_consume.go b/std/object/client_consume.go index 64ba3e3e..78262002 100644 --- a/std/object/client_consume.go +++ b/std/object/client_consume.go @@ -119,7 +119,7 @@ func (c *Client) fetchMetadata( MustBeFresh: true, Lifetime: optional.Some(time.Millisecond * 1000), }, - Retries: 3, // TODO: configurable + Retries: 3, TryStore: utils.If(tryStore, c.store, nil), Callback: func(args ndn.ExpressCallbackArgs) { if args.Result == ndn.InterestResultError { @@ -176,7 +176,7 @@ func (c *Client) fetchDataByPrefix( MustBeFresh: true, Lifetime: optional.Some(time.Millisecond * 1000), }, - Retries: 3, // TODO: configurable + Retries: 3, TryStore: utils.If(tryStore, c.store, nil), Callback: func(args ndn.ExpressCallbackArgs) { if args.Result == ndn.InterestResultError { diff --git a/std/object/client_consume_seg.go b/std/object/client_consume_seg.go index 8fa9440e..651191ca 100644 --- a/std/object/client_consume_seg.go +++ b/std/object/client_consume_seg.go @@ -46,17 +46,22 @@ type retxEntry struct { retries int } -// (AI GENERATED DESCRIPTION): Initializes a new `rrSegFetcher` with the given client, an empty stream list, a fixed congestion window of size 100, no outstanding packets, an empty retransmission queue, a retry counter map, and a maximum of 3 retries. +// Initializes a new rrSegFetcher with the given client. The default congestion +// window is AIMD: it grows by 1 segment on every data arrival and is halved +// on loss or Nack-Congestion. Without AIMD (e.g. FixedCongestionWindow) the +// 53-node sprint e2e topology flaps: many concurrent consumers maintain 100 +// outstanding Interests each, the network becomes congested, and a single +// segment loses 3 retries in a row, aborting the cat fetch. func newRrSegFetcher(client *Client) rrSegFetcher { return rrSegFetcher{ mutex: sync.RWMutex{}, client: client, streams: make([]*ConsumeState, 0), - window: cong.NewFixedCongestionWindow(100), + window: cong.NewAIMDCongestionWindow(100), outstanding: 0, retxQueue: list.New(), txCounter: make(map[*ConsumeState]int), - maxRetries: 3, + maxRetries: 5, } } @@ -148,7 +153,6 @@ func (s *rrSegFetcher) findWork() *ConsumeState { return state } -// (AI GENERATED DESCRIPTION): Checks for pending or retransmitted segments, builds and expresses Interest packets for them (updating the congestion window and retry count), and handles the outcome until no more work or the window becomes full. func (s *rrSegFetcher) check() { for { log.Debug(nil, "Checking for work") @@ -265,6 +269,13 @@ func (s *rrSegFetcher) handleResult(args ndn.ExpressCallbackArgs, state *Consume // congestion signal s.window.HandleSignal(cong.SigCongest) s.enqueueForRetransmission(state, seg, retries-1) + case spec.NackReasonNoRoute: + // Transient: route not yet learned (e.g. DV startup race). + // Treat as a loss so the AIMD window shrinks; the segment + // will be retried until the retry budget is exhausted or a + // route arrives. + s.window.HandleSignal(cong.SigLoss) + s.enqueueForRetransmission(state, seg, retries-1) default: // treat as irrecoverable error for now state.finalizeError(fmt.Errorf("%w: fetch seg failed with result: %s", ndn.ErrNetwork, args.Result)) @@ -300,7 +311,6 @@ func (s *rrSegFetcher) handleData(args ndn.ExpressCallbackArgs, state *ConsumeSt }) } -// (AI GENERATED DESCRIPTION): Handles a validated Data packet by extracting its segment number, storing its payload in the consume state’s buffer, updating counters and sliding windows, and finalizing the state when all segments have been received. func (s *rrSegFetcher) handleValidatedData(args ndn.ExpressCallbackArgs, state *ConsumeState) { // get the final block id if we don't know the segment count if state.segCnt == -1 { // TODO: can change? @@ -398,21 +408,18 @@ func (s *rrSegFetcher) enqueueForRetransmission(state *ConsumeState, seg uint64, s.retxQueue.PushBack(&retxEntry{state, seg, retries}) } -// (AI GENERATED DESCRIPTION): Increments the thread‑safe count of outstanding segment fetch requests. func (s *rrSegFetcher) incrementOutstanding() { s.mutex.Lock() defer s.mutex.Unlock() s.outstanding++ } -// (AI GENERATED DESCRIPTION): Decrements the rrSegFetcher’s outstanding request counter in a thread‑safe manner. func (s *rrSegFetcher) decrementOutstanding() { s.mutex.Lock() defer s.mutex.Unlock() s.outstanding-- } -// (AI GENERATED DESCRIPTION): Decrements the outstanding transmission counter for a given consume state, protecting the update with the fetcher’s mutex. func (s *rrSegFetcher) decrementTxCounter(state *ConsumeState) { s.mutex.Lock() defer s.mutex.Unlock() diff --git a/std/object/expressr.go b/std/object/expressr.go index 10e56b59..5a4f284a 100644 --- a/std/object/expressr.go +++ b/std/object/expressr.go @@ -60,24 +60,36 @@ func ExpressR(engine ndn.Engine, args ndn.ExpressRArgs) { // Send the interest // TODO: reexpress faster than lifetime err = engine.Express(interest, func(res ndn.ExpressCallbackArgs) { - if res.Result == ndn.InterestResultTimeout { + // Retryable transient results: timeout and the Nacks that signal + // "try again later" (no FIB entry yet, or downstream congestion). + // Routing convergence in a large network can outlast a single + // Interest lifetime, especially when DV/NLSR startup produces a + // burst of prefix resets; without retry the first Interest after a + // scenario transition races ahead of route propagation and the + // caller observes an immediate failure. + retryable := false + switch res.Result { + case ndn.InterestResultTimeout: log.Debug(nil, "ExpressR Interest timeout", "name", args.Name) - - // Check if retries are exhausted - if args.Retries == 0 { - args.Callback(res) - return + retryable = true + case ndn.InterestResultNack: + switch res.NackReason { + case spec.NackReasonNoRoute, spec.NackReasonCongestion: + log.Debug(nil, "ExpressR retryable Nack", "name", args.Name, "reason", res.NackReason) + retryable = true } - - // Retry on timeout - args.Retries-- - ExpressR(engine, args) + } + if !retryable { + args.Callback(res) return - } else { - // All other results / errors are final + } + + if args.Retries == 0 { args.Callback(res) return } + args.Retries-- + ExpressR(engine, args) }) if err != nil { finalizeError(err) diff --git a/std/sync/svs.go b/std/sync/svs.go index 60e71015..24940873 100644 --- a/std/sync/svs.go +++ b/std/sync/svs.go @@ -4,6 +4,7 @@ import ( "fmt" "math" rand "math/rand/v2" + "slices" "sync" "sync/atomic" "time" @@ -12,11 +13,17 @@ import ( "github.com/named-data/ndnd/std/log" "github.com/named-data/ndnd/std/ndn" spec "github.com/named-data/ndnd/std/ndn/spec_2022" - spec_svs "github.com/named-data/ndnd/std/ndn/svs/v3" + spec_svs "github.com/named-data/ndnd/std/ndn/svs/v4" "github.com/named-data/ndnd/std/types/optional" "github.com/named-data/ndnd/std/utils" ) +// syncVectorThreshold is the max embedded SvsData size (bytes) above +// which the sender switches to PARTIAL (on publication) or publish+pull +// (on periodic sync and recovery). SVS v4 always emits `mhash` and one of +// FullStateVector/PartialStateVector on the wire (see std/ndn/svs/v4). +const syncVectorThreshold = 1200 + type SvSync struct { o SvSyncOpts @@ -40,6 +47,13 @@ type SvSync struct { // Channel for incoming state vectors recvSv chan svSyncRecvSvArgs + // Prefix for published full State Vector Data (.../32=sv). + fullVectorPrefix enc.Name + + // lastPullTime debounces pullFullVector per sender so a sync storm across + // many peers does not generate thousands of redundant segment-0 fetches. + lastPullTime map[string]time.Time + // cancellation for face hook faceCancel func() } @@ -58,6 +72,13 @@ type SvSyncOpts struct { // If not provided, the GroupPrefix will be used instead. SyncDataName enc.Name + // FullVectorPrefix is the publish/serve prefix for retrievable FULL + // StateVector Data used by SvsDataRef publish+pull recovery. The + // version component is appended when producing the Data. + // If not provided, it defaults to SyncDataName with the trailing + // "32=svs" component (if present) replaced by "32=sv". + FullVectorPrefix enc.Name + // Initial state vector from persistence InitialState *spec_svs.StateVector // Boot time from persistence @@ -83,8 +104,11 @@ type SvSyncUpdate struct { } type svSyncRecvSvArgs struct { - sv *spec_svs.StateVector - data enc.Wire + sv *spec_svs.StateVector + data enc.Wire + partial bool + mhash []byte + svsDataRef enc.Name } // NewSvSync creates a new SV Sync instance. @@ -135,7 +159,7 @@ func NewSvSync(opts SvSyncOpts) *SvSync { mutex: sync.Mutex{}, state: initialState, mtime: make(map[string]time.Time), - prefix: opts.GroupPrefix.Append(enc.NewVersionComponent(3)), + prefix: opts.GroupPrefix.Append(enc.NewVersionComponent(4)), suppress: false, merge: NewSvMap[uint64](0), @@ -145,6 +169,10 @@ func NewSvSync(opts SvSyncOpts) *SvSync { recvSv: make(chan svSyncRecvSvArgs, 128), + fullVectorPrefix: resolveFullVectorPrefix(opts.FullVectorPrefix, opts.SyncDataName), + + lastPullTime: make(map[string]time.Time), + faceCancel: func() {}, } } @@ -169,7 +197,6 @@ func (s *SvSync) Start() (err error) { return nil } -// (AI GENERATED DESCRIPTION): Runs the SvSync event loop: it performs the initial sync (or passive load), registers periodic timer ticks and face‑up callbacks, processes received state vectors, and exits cleanly when signalled to stop. func (s *SvSync) main() { // Cleanup on exit defer s.o.Client.Engine().DetachHandler(s.prefix) @@ -180,7 +207,7 @@ func (s *SvSync) main() { // Notify everyone when we are back online s.faceCancel = s.o.Client.Engine().Face().OnUp(func() { - time.AfterFunc(100*time.Millisecond, s.sendSyncInterest) + time.AfterFunc(100*time.Millisecond, func() { s.sendSyncInterest(syncSendOther) }) }) defer s.faceCancel() @@ -190,7 +217,7 @@ func (s *SvSync) main() { go s.loadPassiveWires() } else { // Send the initial Sync Interest - go s.sendSyncInterest() + go s.sendSyncInterest(syncSendOther) } for { @@ -242,7 +269,7 @@ func (s *SvSync) SetSeqNo(name enc.Name, seqNo uint64) error { // [Spec] When the node generates a new publication, // immediately emit a Sync Interest s.state.Set(hash, s.o.BootTime, seqNo) - go s.sendSyncInterest() + go s.sendSyncInterest(syncSendPublication, name) return nil } @@ -264,17 +291,15 @@ func (s *SvSync) IncrSeqNo(name enc.Name) uint64 { // [Spec] When the node generates a new publication, // immediately emit a Sync Interest - go s.sendSyncInterest() + go s.sendSyncInterest(syncSendPublication, name) return entry } -// (AI GENERATED DESCRIPTION): Returns the boot time value stored in the SvSync instance. func (s *SvSync) GetBootTime() uint64 { return s.o.BootTime } -// (AI GENERATED DESCRIPTION): Returns a thread‑safe slice of all names currently stored in the SvSync state. func (s *SvSync) GetNames() []enc.Name { s.mutex.Lock() defer s.mutex.Unlock() @@ -287,7 +312,6 @@ func (s *SvSync) GetNames() []enc.Name { return names } -// (AI GENERATED DESCRIPTION): Processes an incoming state vector, updating the local state vector, notifying the application of any changes, and handling suppression and passive‑sync logic while ensuring updates are delivered in order. func (s *SvSync) onReceiveStateVector(args svSyncRecvSvArgs) { // Deliver the updates after this call is done // This ensures the mutex is not held during the callback @@ -372,7 +396,22 @@ func (s *SvSync) onReceiveStateVector(args svSyncRecvSvArgs) { // The above checks each node in the incoming state vector, but // does not check if a node is missing from the incoming state vector. - if !isOutdated && s.state.IsNewerThan(recvSv, func(_, _ uint64) bool { return false }) { + // + // [Spec] The wire TLV (FullStateVector or PartialStateVector) replaces + // the previous VectorType discriminator. publish-only Sync Data carries + // no StateVector and is filtered out earlier (see onSyncData), so any + // args.sv reaching this function was either FULL or PARTIAL on the wire. + isPartial := args.partial + // [Spec] Membership recovery is a FULL-boundary operation: only an embedded + // FULL StateVector or a publish-only Sync Data (which carries no StateVector + // and is filtered earlier in onSyncData) represents the sender's complete + // membership view. A PARTIAL vector is a subset by design, so the recvSv + // tuple-count superset check in handleMhashMismatch would spuriously + // trigger sendRecoveryPublish for the local node's normal PUBLISH path. + if len(args.mhash) > 0 && !isPartial { + s.handleMhashMismatch(args, recvSv) + } + if !isPartial && !isOutdated && s.state.IsNewerThan(recvSv, func(_, _ uint64) bool { return false }) { isOutdated = true canDrop = false } @@ -402,7 +441,6 @@ func (s *SvSync) onReceiveStateVector(args svSyncRecvSvArgs) { s.ticker.Reset(s.getSuppressionTimeout()) } -// (AI GENERATED DESCRIPTION): Handles a timer expiry by checking suppression state, potentially transitioning to steady state, and asynchronously sending a Sync Interest with the current local state vector. func (s *SvSync) timerExpired() { s.mutex.Lock() defer s.mutex.Unlock() @@ -420,11 +458,10 @@ func (s *SvSync) timerExpired() { // [Spec] On expiration of timer emit a Sync Interest // with the current local state vector. - go s.sendSyncInterest() + go s.sendSyncInterest(syncSendPeriodic) } -// (AI GENERATED DESCRIPTION): Sends a sync Interest: if passive mode is enabled, it publishes all buffered state updates without duplicates; otherwise, it encodes the current state vector into a wire and transmits it, provided the sync service is running. -func (s *SvSync) sendSyncInterest() { +func (s *SvSync) sendSyncInterest(reason syncSendReason, pubName ...enc.Name) { if !s.running.Load() { return } @@ -437,12 +474,16 @@ func (s *SvSync) sendSyncInterest() { return } + var sender enc.Name + if reason == syncSendPublication && len(pubName) > 0 { + sender = pubName[0] + } + // Encode and sign the current state vector - wire := s.encodeSyncData() + wire := s.encodeSyncData(reason, sender) s.sendSyncInterestWith(wire) } -// (AI GENERATED DESCRIPTION): Sends a sync Interest carrying the supplied data wire payload with a 1‑second lifetime, using the object’s prefix, and logs any construction or transmission errors. func (s *SvSync) sendSyncInterestWith(dataWire enc.Wire) { if dataWire == nil { return @@ -465,21 +506,58 @@ func (s *SvSync) sendSyncInterestWith(dataWire enc.Wire) { } } -// (AI GENERATED DESCRIPTION): Builds a signed Data packet containing the current state vector for SVS v3 synchronization. -func (s *SvSync) encodeSyncData() enc.Wire { - // Critical section - sv := func() *spec_svs.StateVector { - s.mutex.Lock() - defer s.mutex.Unlock() - - // [Spec*] Sending always triggers Steady State - s.enterSteadyState() - - return s.state.Encode(func(s uint64) uint64 { return s }) - }() - svWire := (&spec_svs.SvsData{StateVector: sv}).Encode() +func (s *SvSync) encodeSyncData(reason syncSendReason, sender enc.Name) enc.Wire { + s.mutex.Lock() + s.enterSteadyState() + stateSnap := cloneSvMap(s.state) + mtimeSnap := make(map[string]time.Time, len(s.mtime)) + for k, v := range s.mtime { + mtimeSnap[k] = v + } + repair, propagation := s.partialTargets() + s.mutex.Unlock() + + var svsData *spec_svs.SvsData + usePublish, precomputedFull, _ := shouldUsePublishPull(reason, syncVectorThreshold, stateSnap) + switch { + case usePublish: + ref, err := s.publishFullVectorData(stateSnap) + if err != nil { + log.Error(s, "publishFullVectorData failed", "err", err) + return nil + } + svsData = buildPublishSvsData(stateSnap, ref) + case precomputedFull != nil: + // [Spec] Inline FULL fits the threshold; reuse the SvsData we + // already built during the size check instead of re-encoding. + svsData = precomputedFull + default: + svsData = buildSvsDataForSend(svsSendInput{ + State: stateSnap, + Reason: reason, + Threshold: syncVectorThreshold, + Sender: sender, + Repair: repair, + Propagation: propagation, + Mtime: mtimeSnap, + }) + if svsData == nil { + // [Spec] Publication-triggered PARTIAL encoding could not fit + // even the sender-only baseline: fall back to publish+pull. + ref, err := s.publishFullVectorData(stateSnap) + if err != nil { + log.Error(s, "publishFullVectorData failed (fallback)", "err", err) + return nil + } + svsData = buildPublishSvsData(stateSnap, ref) + } + } + if svsData == nil { + return nil + } + svWire := svsData.Encode() - // SVS v3 Sync Data + // SVS v4 Sync Data name := s.o.SyncDataName.WithVersion(enc.VersionUnixMicro) // Sign Sync Data @@ -500,7 +578,6 @@ func (s *SvSync) encodeSyncData() enc.Wire { return data.Wire } -// (AI GENERATED DESCRIPTION): Handles a received sync Interest by checking the running state, extracting its AppParam, and passing that payload to the sync‑data processing routine. func (s *SvSync) onSyncInterest(interest ndn.Interest) { if !s.running.Load() { return @@ -516,7 +593,6 @@ func (s *SvSync) onSyncInterest(interest ndn.Interest) { s.onSyncData(interest.AppParam()) } -// (AI GENERATED DESCRIPTION): Processes a received SyncData packet by parsing it, validating the signature, extracting the state vector, and forwarding the vector and original data to the receiver channel. func (s *SvSync) onSyncData(dataWire enc.Wire) { data, sigCov, err := spec.Spec{}.ReadData(enc.NewWireView(dataWire)) if err != nil { @@ -536,17 +612,54 @@ func (s *SvSync) onSyncData(dataWire enc.Wire) { return } - // Decode state vector + // Decode SvsData (embedded FULL, embedded PARTIAL, or publish-only ref). svWire := data.Content().Join() params, err := spec_svs.ParseSvsData(enc.NewBufferView(svWire), false) - if err != nil || params.StateVector == nil { - log.Warn(s, "onSyncInterest failed to parse StateVec", "err", err) + if err != nil { + log.Warn(s, "onSyncInterest failed to parse SvsData", "err", err) + return + } + + mhash := params.MemberSetHash + sv := params.GetStateVector() + + // Publish-only ref: advertise that the full vector is retrievable. + if sv == nil && len(params.SvsDataRef) > 0 { + // [Spec] Every Sync Data carries a 32-byte mhash. Reject malformed + // packets that omit mhash on the publish-only form. + if len(mhash) != 32 { + log.Warn(s, "onSyncInterest publish-only SvsData missing or invalid mhash", + "len", len(mhash)) + return + } + trustPrefix := pullRefFromSyncDataWire(dataWire) + go s.pullFullVector(params.SvsDataRef, trustPrefix) + return + } + if sv == nil { + log.Warn(s, "onSyncInterest SvsData has no StateVector") + return + } + + // [Spec] Direct form must carry exactly one of FullStateVector / + // PartialStateVector, and a 32-byte mhash. The wire TLV replaces + // the previous VectorType discriminator. + if params.IsFull() == params.IsPartial() || len(params.SvsDataRef) > 0 { + log.Warn(s, "onSyncInterest inline SvsData has invalid direct form") + return + } + if len(mhash) != 32 { + log.Warn(s, "onSyncInterest inline SvsData missing or invalid mhash", + "len", len(mhash)) return } s.recvSv <- svSyncRecvSvArgs{ - sv: params.StateVector, - data: dataWire, + sv: sv, + data: dataWire, + partial: params.IsPartial(), + mhash: mhash, + svsDataRef: params.SvsDataRef, } }, }) @@ -559,7 +672,6 @@ func (s *SvSync) enterSteadyState() { s.ticker.Reset(s.getPeriodicTimeout()) } -// (AI GENERATED DESCRIPTION): Returns a duration uniformly randomized within ±10% of the configured periodic timeout. func (s *SvSync) getPeriodicTimeout() time.Duration { // [Spec] ±10% uniform jitter jitter := s.o.PeriodicTimeout / 10 @@ -568,7 +680,6 @@ func (s *SvSync) getPeriodicTimeout() time.Duration { return time.Duration(rand.Int64N(int64(max-min))) + min } -// (AI GENERATED DESCRIPTION): Calculates a random suppression timeout duration using an exponential‑decay function based on the configured SuppressionPeriod. func (s *SvSync) getSuppressionTimeout() time.Duration { // [Spec] Exponential decay function // [Spec] c = SuppressionPeriod // constant factor @@ -664,5 +775,19 @@ func (s *SvSync) loadPassiveWires() { } // This is hacky but pragmatic - wait for the state to be processed - time.AfterFunc(500*time.Millisecond, s.sendSyncInterest) + time.AfterFunc(500*time.Millisecond, func() { s.sendSyncInterest(syncSendOther) }) +} + +// partialTargets returns the repair target names from the suppression-merge +// state, sorted in NDN canonical order so PARTIAL selection is deterministic. +// propagation is currently unused and always nil. +func (s *SvSync) partialTargets() (repair, propagation []enc.Name) { + if !s.suppress { + return nil, nil + } + for name := range s.merge.Iter() { + repair = append(repair, name) + } + slices.SortFunc(repair, func(a, b enc.Name) int { return a.Compare(b) }) + return repair, nil } diff --git a/std/sync/svs_alo_data.go b/std/sync/svs_alo_data.go index 27543385..7be450b8 100644 --- a/std/sync/svs_alo_data.go +++ b/std/sync/svs_alo_data.go @@ -25,7 +25,6 @@ type svsDataState struct { SnapBlock int } -// (AI GENERATED DESCRIPTION): Builds the full name for a data object by appending the node identifier, boot‑timestamp, and sequence number to the group’s prefix and marking the resulting name as immutable. func (s *SvsALO) objectName(node enc.Name, boot uint64, seq uint64) enc.Name { return s.GroupPrefix(). Append(node...). @@ -34,7 +33,6 @@ func (s *SvsALO) objectName(node enc.Name, boot uint64, seq uint64) enc.Name { WithVersion(enc.VersionImmutable) } -// (AI GENERATED DESCRIPTION): Publishes a new Data object with the supplied content, updates the SVS state vector and snapshot strategy, and returns the produced name and the instance’s serialized state. func (s *SvsALO) produceObject(content enc.Wire) (enc.Name, enc.Wire, error) { // This instance owns the underlying SVS instance. // So we can be sure that the sequence number does not @@ -118,6 +116,8 @@ func (s *SvsALO) consumeObject(node enc.Name, boot uint64, seq uint64) { fetchName := s.objectName(node, boot, seq) s.client.ConsumeExt(ndn.ConsumeExtArgs{ Name: fetchName, + TryStore: true, + NoMetadata: true, // fetch name includes version+seq; metadata would only block on timeout UseSignatureTime: s.opts.Svs.UseSignatureTime, IgnoreValidity: s.opts.Svs.IgnoreValidity, Callback: func(status ndn.ConsumeState) { diff --git a/std/sync/svs_encode.go b/std/sync/svs_encode.go new file mode 100644 index 00000000..2db361a5 --- /dev/null +++ b/std/sync/svs_encode.go @@ -0,0 +1,263 @@ +package sync + +import ( + "cmp" + "math/rand/v2" + "slices" + "time" + + enc "github.com/named-data/ndnd/std/encoding" + spec_svs "github.com/named-data/ndnd/std/ndn/svs/v4" +) + +// syncSendReason distinguishes why a Sync Interest is being sent. +type syncSendReason int + +const ( + syncSendOther syncSendReason = iota + syncSendPublication + syncSendPeriodic + syncSendRecovery +) + +// PartialEncodeOpts configures subset selection for inline PARTIAL vectors. +type PartialEncodeOpts struct { + Sender enc.Name + Threshold int + Repair []enc.Name + Propagation []enc.Name + Mtime map[string]time.Time +} + +// svsSendInput carries everything needed to build inline Sync Data for send. +type svsSendInput struct { + State SvMap[uint64] + Reason syncSendReason + Threshold int + Sender enc.Name + Repair []enc.Name + Propagation []enc.Name + Mtime map[string]time.Time +} + +// buildSvsDataForSend picks embedded FULL or PARTIAL SvsData for an outgoing +// Sync message. Returns nil when publication-triggered PARTIAL encoding cannot +// fit even the sender-only baseline (signaled by an empty StateVector from +// encodePartialStateVector): the caller MUST fall back to publish+pull +// (see shouldUsePublishPull). Other reasons always return a non-nil result. +func buildSvsDataForSend(in svsSendInput) *spec_svs.SvsData { + fullSv := in.State.Encode(func(seq uint64) uint64 { return seq }) + fullData := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(in.State), + FullStateVector: &spec_svs.FullStateVector{StateVector: fullSv}, + } + + if in.Reason != syncSendPublication || len(fullData.Encode().Join()) <= in.Threshold { + return fullData + } + + partialSv := encodePartialStateVector(in.State, PartialEncodeOpts{ + Sender: in.Sender, + Threshold: in.Threshold, + Repair: in.Repair, + Propagation: in.Propagation, + Mtime: in.Mtime, + }) + if len(partialSv.Entries) == 0 { + // Baseline exceeded Threshold; caller must use publish+pull. + return nil + } + return &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(in.State), + PartialStateVector: &spec_svs.PartialStateVector{StateVector: partialSv}, + } +} + +// encodePartialStateVector builds a PARTIAL StateVector for new publication. +// Entry [0] is the sender; entries [1..n] are in NDN canonical order. +// +// Returns an empty StateVector (no entries) if the sender-only baseline +// itself exceeds Threshold: the caller MUST fall back to publish+pull in +// that case and the empty vector is the explicit signal that no PARTIAL +// body could be fit under the size budget. +func encodePartialStateVector(state SvMap[uint64], opts PartialEncodeOpts) *spec_svs.StateVector { + seq := func(v uint64) uint64 { return v } + senderHash := opts.Sender.TlvStr() + mhash := ComputeMembershipHash(state) + + senderEntry := state.encodeNameEntry(opts.Sender, seq) + if senderEntry == nil { + senderEntry = &spec_svs.StateVectorEntry{Name: opts.Sender} + } + + // Sender-only baseline must always fit when possible. + baseline := &spec_svs.StateVector{Entries: []*spec_svs.StateVectorEntry{senderEntry}} + baselineData := &spec_svs.SvsData{ + MemberSetHash: mhash, + PartialStateVector: &spec_svs.PartialStateVector{StateVector: baseline}, + } + if len(baselineData.Encode().Join()) > opts.Threshold { + // Baseline too large to fit even the sender entry: return an + // empty PARTIAL so the caller can detect the overflow and fall + // back to publish+pull. + return &spec_svs.StateVector{} + } + + peers := priorityOrderedPeers(state, senderHash, opts) + included := map[string]bool{senderHash: true} + entries := []*spec_svs.StateVectorEntry{senderEntry} + + for _, name := range peers { + hash := name.TlvStr() + if included[hash] { + continue + } + entry := state.encodeNameEntry(name, seq) + if entry == nil { + continue + } + + trial := append(slices.Clone(entries), entry) + sortPartialTail(trial) + trialSv := &spec_svs.StateVector{Entries: trial} + trialData := &spec_svs.SvsData{ + MemberSetHash: mhash, + PartialStateVector: &spec_svs.PartialStateVector{StateVector: trialSv}, + } + if len(trialData.Encode().Join()) > opts.Threshold { + break + } + + entries = trial + included[hash] = true + } + + // entries is already sorted tail-wise: the loop's sortPartialTail(trial) + // runs on every accepted iteration, so the trailing sort is redundant. + return &spec_svs.StateVector{Entries: entries} +} + +// priorityOrderedPeers returns the producer names that consumers should +// consider for inclusion in a new-publication PARTIAL StateVector, ordered +// by inclusion priority. The caller iterates the returned slice and +// greedily adds entries until the size budget is reached. +// +// The priority order is: +// +// 1. Repair targets from the suppression-merge state (newest entries first). +// 2. Propagation targets (the most recently updated producers). +// 3. Inactive producers (zero-value entries) in randomized order — these +// are included only when bandwidth allows, so randomization is fair. +// 4. Remaining active producers, sorted by (a) recency descending then +// (b) canonical NDN name ascending. +// +// The sender is excluded — it is always included at entries[0]. +func priorityOrderedPeers(state SvMap[uint64], senderHash string, opts PartialEncodeOpts) []enc.Name { + seen := map[string]bool{senderHash: true} + out := make([]enc.Name, 0, len(state)) + + appendUnique := func(names []enc.Name) { + for _, name := range names { + hash := name.TlvStr() + if seen[hash] { + continue + } + if _, ok := state[hash]; !ok { + continue + } + seen[hash] = true + out = append(out, name) + } + } + + appendUnique(opts.Repair) + appendUnique(opts.Propagation) + + inactive := make([]enc.Name, 0) + remaining := make([]enc.Name, 0) + for name, vals := range state.Iter() { + hash := name.TlvStr() + if seen[hash] { + continue + } + if isInactiveProducer(vals) { + inactive = append(inactive, name) + continue + } + remaining = append(remaining, name) + } + + rand.Shuffle(len(inactive), func(i, j int) { + inactive[i], inactive[j] = inactive[j], inactive[i] + }) + appendUnique(inactive) + + // Sort by (recency desc, canonical name asc) using a single comparator. + // slices.SortFunc is not stable, so sorting twice is not a reliable + // tie-breaker for entries with equal recency scores. + slices.SortFunc(remaining, func(a, b enc.Name) int { + if c := cmp.Compare(recencyScore(opts.Mtime, b), recencyScore(opts.Mtime, a)); c != 0 { + return c + } + return a.Compare(b) + }) + appendUnique(remaining) + + return out +} + +func isInactiveProducer(vals []SvMapVal[uint64]) bool { + for _, val := range vals { + if val.Value > 0 { + return false + } + } + return true +} + +func recencyScore(mtime map[string]time.Time, name enc.Name) int64 { + if mtime == nil { + return 0 + } + t, ok := mtime[name.TlvStr()] + if !ok { + return 0 + } + return t.UnixNano() +} + +// sortPartialTail keeps the first entry fixed (the sender) and sorts the +// remaining entries in canonical NDN name order. PARTIAL vectors place the +// sender at entries[0] and the rest must be canonically ordered for byte- +// deterministic encoding. +func sortPartialTail(entries []*spec_svs.StateVectorEntry) { + if len(entries) <= 1 { + return + } + slices.SortFunc(entries[1:], func(a, b *spec_svs.StateVectorEntry) int { + return a.Name.Compare(b.Name) + }) +} + +// encodeNameEntry encodes one producer name from the map. +func (m SvMap[V]) encodeNameEntry(name enc.Name, seq func(V) uint64) *spec_svs.StateVectorEntry { + hash := name.TlvStr() + vals, ok := m[hash] + if !ok { + return nil + } + + entry := &spec_svs.StateVectorEntry{ + Name: name, + SeqNoEntries: make([]*spec_svs.SeqNoEntry, 0, len(vals)), + } + for _, val := range vals { + if seqNo := seq(val.Value); seqNo > 0 { + entry.SeqNoEntries = append(entry.SeqNoEntries, &spec_svs.SeqNoEntry{ + BootstrapTime: val.Boot, + SeqNo: seqNo, + }) + } + } + return entry +} diff --git a/std/sync/svs_map.go b/std/sync/svs_map.go index be1ffd0c..f19be637 100644 --- a/std/sync/svs_map.go +++ b/std/sync/svs_map.go @@ -7,7 +7,7 @@ import ( enc "github.com/named-data/ndnd/std/encoding" "github.com/named-data/ndnd/std/log" - spec_svs "github.com/named-data/ndnd/std/ndn/svs/v3" + spec_svs "github.com/named-data/ndnd/std/ndn/svs/v4" ) // Map representation of the state vector. @@ -19,7 +19,6 @@ type SvMapVal[V any] struct { Value V } -// (AI GENERATED DESCRIPTION): Compares the Boot field of two SvMapVal[V] values, returning a negative, zero, or positive integer to indicate their ordering. func (*SvMapVal[V]) Cmp(a, b SvMapVal[V]) int { return cmp.Compare(a.Boot, b.Boot) } @@ -29,6 +28,15 @@ func NewSvMap[V any](size int) SvMap[V] { return make(SvMap[V], size) } +// cloneSvMap returns a shallow copy safe for use without holding SvSync.mutex. +func cloneSvMap[V any](m SvMap[V]) SvMap[V] { + out := NewSvMap[V](len(m)) + for hash, vals := range m { + out[hash] = slices.Clone(vals) + } + return out +} + // Get seq entry for a bootstrap time. func (m SvMap[V]) Get(hash string, boot uint64) (value V) { entry := SvMapVal[V]{boot, value} @@ -39,7 +47,6 @@ func (m SvMap[V]) Get(hash string, boot uint64) (value V) { return value } -// (AI GENERATED DESCRIPTION): Adds or updates a value in the sorted list for a given hash, inserting the new entry or replacing the existing one while maintaining the slice sorted by the boot field. func (m SvMap[V]) Set(hash string, boot uint64, value V) { entry := SvMapVal[V]{boot, value} i, match := slices.BinarySearchFunc(m[hash], entry, entry.Cmp) @@ -50,7 +57,6 @@ func (m SvMap[V]) Set(hash string, boot uint64, value V) { m[hash] = slices.Insert(m[hash], i, entry) } -// (AI GENERATED DESCRIPTION): Clears all key/value pairs from the SvMap, safely handling nil maps by doing nothing if the map is nil. func (m SvMap[V]) Clear() { if m != nil { clear(m) @@ -118,7 +124,6 @@ func (m SvMap[V]) Encode(seq func(V) uint64) *spec_svs.StateVector { return &spec_svs.StateVector{Entries: entries} } -// (AI GENERATED DESCRIPTION): Iter returns an iterator over the SvMap that yields each decoded name and its associated slice of SvMapVal values. func (m SvMap[V]) Iter() iter.Seq2[enc.Name, []SvMapVal[V]] { return func(yield func(enc.Name, []SvMapVal[V]) bool) { for hash, val := range m { diff --git a/std/sync/svs_membership_hash.go b/std/sync/svs_membership_hash.go new file mode 100644 index 00000000..d0aae4d3 --- /dev/null +++ b/std/sync/svs_membership_hash.go @@ -0,0 +1,42 @@ +package sync + +import ( + "crypto/sha256" + "slices" + + spec_svs "github.com/named-data/ndnd/std/ndn/svs/v4" +) + +// ComputeMembershipHash returns the membership hash over all (Name, BootstrapTime) pairs in state. +// Each tuple is encoded as a TLV structure (Tuple-T 0xcc with Name and BootstrapTime +// children) using the ndnd standard TLV codec. +func ComputeMembershipHash(state SvMap[uint64]) []byte { + membershipTuples := make([]*spec_svs.MembershipTuple, 0) + for name, vals := range state.Iter() { + for _, val := range vals { + membershipTuples = append(membershipTuples, &spec_svs.MembershipTuple{ + Name: name, + BootstrapTime: val.Boot, + }) + } + } + + slices.SortFunc(membershipTuples, func(a, b *spec_svs.MembershipTuple) int { + if c := a.Name.Compare(b.Name); c != 0 { + return c + } + if a.BootstrapTime < b.BootstrapTime { + return -1 + } + if a.BootstrapTime > b.BootstrapTime { + return 1 + } + return 0 + }) + + h := sha256.New() + for _, t := range membershipTuples { + h.Write(t.Encode().Join()) + } + return h.Sum(nil) +} diff --git a/std/sync/svs_pull.go b/std/sync/svs_pull.go new file mode 100644 index 00000000..dd37a2df --- /dev/null +++ b/std/sync/svs_pull.go @@ -0,0 +1,298 @@ +package sync + +import ( + "bytes" + "fmt" + "time" + + enc "github.com/named-data/ndnd/std/encoding" + "github.com/named-data/ndnd/std/log" + "github.com/named-data/ndnd/std/ndn" + spec "github.com/named-data/ndnd/std/ndn/spec_2022" + spec_svs "github.com/named-data/ndnd/std/ndn/svs/v4" +) + +const ( + syncDataKeyword = "svs" + fullVectorKeyword = "sv" +) + +// deriveFullVectorPrefix is an implementation convenience that maps a +// SyncDataName (.../32=svs) to the published full-vector prefix +// (.../32=sv) by replacing the trailing keyword. The spec only requires +// that a full-vector prefix exists at some sender-controlled location; this +// helper just provides a default when the caller does not set one. +// Callers that wire to a different prefix MUST supply FullVectorPrefix +// explicitly via NewSvSync opts. +func deriveFullVectorPrefix(syncDataName enc.Name) enc.Name { + if len(syncDataName) == 0 { + return nil + } + base := syncDataName + if base.At(-1).IsKeyword(syncDataKeyword) { + base = base.Prefix(-1) + } + return base.Append(enc.NewKeywordComponent(fullVectorKeyword)) +} + +// resolveFullVectorPrefix returns the explicit FullVectorPrefix if set, +// otherwise derives it from SyncDataName. Per the spec, the full-vector +// prefix is whatever location the producer publishes at; this helper picks +// the default location documented above. +func resolveFullVectorPrefix(explicit, syncDataName enc.Name) enc.Name { + if len(explicit) > 0 { + return explicit.Clone() + } + return deriveFullVectorPrefix(syncDataName) +} + +// pullRefFromSyncDataWire returns the trust prefix for fetching a publish-only +// SvsDataRef. The "ref" field of Sync Data is the published full-vector name +// (.../32=sv/); the trust prefix is the same name with the version +// component stripped, which corresponds to the sender's .../32=sv prefix used +// for all its published full vectors and is what an authorized consumer must +// trust to follow the reference. +func pullRefFromSyncDataWire(dataWire enc.Wire) enc.Name { + data, _, err := spec.Spec{}.ReadData(enc.NewWireView(dataWire)) + if err != nil { + return nil + } + name := data.Name() + if len(name) == 0 { + return nil + } + if name.At(-1).IsVersion() { + name = name.Prefix(-1) + } + return deriveFullVectorPrefix(name) +} + +// buildPublishSvsData constructs the publish-only form of SvsData carried +// in a Sync message (per spec §3.1.2): the membership hash for the current +// state plus a reference (SvsDataRef) to the sender's retrievable full +// vector. The ref is the published full-vector Data name (e.g. +// ////32=sv/); receivers fetch it via +// pullFullVector, validate, and merge. The ref must sit below the +// sender's trust prefix (see isTrustedSvsDataRef). +func buildPublishSvsData(state SvMap[uint64], ref enc.Name) *spec_svs.SvsData { + return &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(state), + SvsDataRef: ref, + } +} + +// shouldUsePublishPull reports whether the sender should publish at .../32=sv +// and emit publish-only Sync Data (SvsDataRef, no embedded vector) instead +// of an embedded FULL or PARTIAL StateVector. The full SvsData is built +// once (so the caller can reuse it if not publishing) and its size is +// returned alongside the decision so the caller does not have to re-encode. +func shouldUsePublishPull(reason syncSendReason, threshold int, state SvMap[uint64]) (usePublish bool, data *spec_svs.SvsData, size int) { + switch reason { + case syncSendRecovery: + return true, nil, 0 + case syncSendPublication: + return false, nil, 0 + } + sv := state.Encode(func(seq uint64) uint64 { return seq }) + full := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(state), + FullStateVector: &spec_svs.FullStateVector{StateVector: sv}, + } + wire := full.Encode().Join() + if len(wire) > threshold { + return true, nil, len(wire) + } + return false, full, len(wire) +} + +// publishFullVectorData produces retrievable inline FULL SvsData at .../32=sv/. +func (s *SvSync) publishFullVectorData(state SvMap[uint64]) (enc.Name, error) { + if len(s.fullVectorPrefix) == 0 { + return nil, fmt.Errorf("full vector prefix unset") + } + sv := state.Encode(func(seq uint64) uint64 { return seq }) + content := (&spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(state), + FullStateVector: &spec_svs.FullStateVector{StateVector: sv}, + }).Encode() + name := s.fullVectorPrefix.WithVersion(enc.VersionUnixMicro) + return s.o.Client.Produce(ndn.ProduceArgs{ + Name: name, + Content: content, + }) +} + +// pullFullVectorMinInterval debounces pullFullVector per sender. During convergence on a +// large group, every sync that crosses the membership hash boundary schedules a pull; without +// gating, a node can accumulate redundant segment-0 fetches for the same content, which +// exhausts retry budgets under network load. We allow at most one pull per sender per +// pullFullVectorMinInterval. +const pullFullVectorMinInterval = 5 * time.Second + +// pullFullVector fetches a published full State Vector and merges it on the main loop. +// trustPrefix is the sender's .../32=sv prefix; ref must be equal to or below it. +// +// [Impl] The 5s per-sender debounce (pullFullVectorMinInterval) is an +// implementation detail documented in §5.6 of the v4 spec: it limits the +// fan-in when many peers cross an mhash boundary at the same time and is +// safe to relax provided the consumer's retry budget scales accordingly. +func (s *SvSync) pullFullVector(ref enc.Name, trustPrefix enc.Name) { + if len(ref) == 0 { + return + } + if !isTrustedSvsDataRef(ref, trustPrefix) { + log.Warn(s, "pullFullVector rejected untrusted SvsDataRef", "ref", ref, "trust", trustPrefix) + return + } + + // Debounce per sender: drop the pull if one is already in flight or completed recently. + senderHash := trustPrefix.TlvStr() + s.mutex.Lock() + if s.lastPullTime == nil { + s.lastPullTime = make(map[string]time.Time) + } + if last, ok := s.lastPullTime[senderHash]; ok && time.Since(last) < pullFullVectorMinInterval { + s.mutex.Unlock() + return + } + s.lastPullTime[senderHash] = time.Now() + s.mutex.Unlock() + + s.o.Client.ConsumeExt(ndn.ConsumeExtArgs{ + Name: ref.Clone(), + TryStore: true, + NoMetadata: true, + UseSignatureTime: s.o.UseSignatureTime, + IgnoreValidity: s.o.IgnoreValidity, + Callback: func(st ndn.ConsumeState) { + if st.Error() != nil { + log.Warn(s, "pullFullVector failed", "ref", ref, "err", st.Error()) + return + } + if !st.IsComplete() { + return + } + s.onPulledFullVector(st.Content().Join()) + }, + }) +} + +// onPulledFullVector merges a fetched inline FULL SvsData into local state. +// Segment signatures are validated by client.ConsumeExt during fetch. +func (s *SvSync) onPulledFullVector(content []byte) { + params, err := parseFullVectorContent(content) + if err != nil { + log.Warn(s, "onPulledFullVector parse failed", "err", err) + return + } + + s.recvSv <- svSyncRecvSvArgs{ + sv: params.GetStateVector(), + mhash: params.MemberSetHash, + } +} + +func parseFullVectorContent(content []byte) (*spec_svs.SvsData, error) { + params, err := spec_svs.ParseSvsData(enc.NewBufferView(content), false) + if err != nil { + return nil, err + } + // [Spec] Fetched full-vector Data is inline FULL: the wire TLV must be + // FullStateVector (0xCD), not PartialStateVector (0xCE), and the + // MemberSetHash must be present and valid. + if !params.IsFull() { + return nil, fmt.Errorf("full vector content is not FullStateVector") + } + sv := params.GetStateVector() + if sv == nil { + return nil, fmt.Errorf("full vector content has no StateVector") + } + mhash := params.MemberSetHash + if len(mhash) != 32 { + return nil, fmt.Errorf("full vector content missing or invalid mhash (len=%d)", len(mhash)) + } + computed := ComputeMembershipHash(stateVectorToMap(sv)) + if !bytes.Equal(mhash, computed) { + return nil, fmt.Errorf("full vector mhash mismatch") + } + return params, nil +} + +func stateVectorToMap(sv *spec_svs.StateVector) SvMap[uint64] { + m := NewSvMap[uint64](len(sv.Entries)) + for _, node := range sv.Entries { + hash := node.Name.TlvStr() + for _, entry := range node.SeqNoEntries { + m.Set(hash, entry.BootstrapTime, entry.SeqNo) + } + } + return m +} + +// sendRecoveryPublish publishes at 32=sv and emits publish-only Sync Data (mhash recovery). +func (s *SvSync) sendRecoveryPublish() { + if !s.running.Load() || s.o.Passive { + return + } + wire := s.encodeSyncData(syncSendRecovery, enc.Name{}) + s.sendSyncInterestWith(wire) +} + +// handleMhashMismatch schedules publish or pull recovery on membership mismatch. +func (s *SvSync) handleMhashMismatch(args svSyncRecvSvArgs, recvSv SvMap[uint64]) { + localMhash := ComputeMembershipHash(s.state) + if bytes.Equal(localMhash, args.mhash) { + return + } + + trustPrefix := pullRefFromSyncDataWire(args.data) + if len(trustPrefix) == 0 { + trustPrefix = s.fullVectorPrefix + } + + localTuples, remoteTuples := membershipTupleCount(s.state), membershipTupleCount(recvSv) + if localTuples > remoteTuples && membershipContains(s.state, recvSv) { + go s.sendRecoveryPublish() + return + } + + // [Spec] Inline FULL is already merged in onReceiveStateVector. + // Pull only when the sender provided a retrievable SvsDataRef (publish-only sync). + if len(args.svsDataRef) == 0 { + return + } + go s.pullFullVector(args.svsDataRef, trustPrefix) +} + +func membershipContains(outer, inner SvMap[uint64]) bool { + for hash, vals := range inner { + for _, v := range vals { + found := false + for _, ov := range outer[hash] { + if ov.Boot == v.Boot { + found = true + break + } + } + if !found { + return false + } + } + } + return true +} + +func membershipTupleCount(m SvMap[uint64]) int { + n := 0 + for _, vals := range m { + n += len(vals) + } + return n +} + +func isTrustedSvsDataRef(ref, senderFullVectorPrefix enc.Name) bool { + if len(ref) == 0 || len(senderFullVectorPrefix) == 0 { + return false + } + return senderFullVectorPrefix.IsPrefix(ref) +} diff --git a/std/sync/svs_test.go b/std/sync/svs_test.go new file mode 100644 index 00000000..26ea675f --- /dev/null +++ b/std/sync/svs_test.go @@ -0,0 +1,512 @@ +package sync + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + + enc "github.com/named-data/ndnd/std/encoding" + "github.com/named-data/ndnd/std/ndn" + spec "github.com/named-data/ndnd/std/ndn/spec_2022" + spec_svs "github.com/named-data/ndnd/std/ndn/svs/v4" + sig "github.com/named-data/ndnd/std/security/signer" + "github.com/named-data/ndnd/std/types/optional" + tu "github.com/named-data/ndnd/std/utils/testutils" +) + +// --- shared test helpers and encode tests --- + +func testSvMapAliceBob() SvMap[uint64] { + m := NewSvMap[uint64](0) + m.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 5) + m.Set(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150, 3) + return m +} + +func TestBuildInlineFullSvsData(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + sv := m.Encode(func(s uint64) uint64 { return s }) + data := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + FullStateVector: &spec_svs.FullStateVector{StateVector: sv}, + } + + require.True(t, data.IsFull()) + require.False(t, data.IsPartial()) + require.Equal(t, ComputeMembershipHash(m), data.MemberSetHash) + svOut := data.GetStateVector() + require.NotNil(t, svOut) + require.Len(t, svOut.Entries, 2) +} + +func TestOnReceivePartialSkipsMissingNameOutdated(t *testing.T) { + tu.SetT(t) + + s := &SvSync{ + o: SvSyncOpts{ + OnUpdate: func(SvSyncUpdate) {}, + SuppressionPeriod: 200 * time.Millisecond, + PeriodicTimeout: 30 * time.Second, + }, + state: testSvMapAliceBob(), + mtime: make(map[string]time.Time), + ticker: time.NewTicker(30 * time.Second), + suppress: false, + } + + // PARTIAL with only bob; local knows alice — must not enter suppression. + bobOnly := NewSvMap[uint64](0) + bobOnly.Set(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150, 3) + partialSv := bobOnly.Encode(func(s uint64) uint64 { return s }) + + s.onReceiveStateVector(svSyncRecvSvArgs{ + sv: partialSv, + partial: true, + mhash: ComputeMembershipHash(bobOnly), + }) + + require.False(t, s.suppress) +} + +func TestOnReceiveFullTreatsMissingNameOutdated(t *testing.T) { + tu.SetT(t) + + s := &SvSync{ + o: SvSyncOpts{ + OnUpdate: func(SvSyncUpdate) {}, + SuppressionPeriod: 200 * time.Millisecond, + PeriodicTimeout: 30 * time.Second, + }, + state: testSvMapAliceBob(), + mtime: make(map[string]time.Time), + ticker: time.NewTicker(30 * time.Second), + suppress: false, + } + + bobOnly := NewSvMap[uint64](0) + bobOnly.Set(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150, 3) + fullSv := bobOnly.Encode(func(s uint64) uint64 { return s }) + + s.onReceiveStateVector(svSyncRecvSvArgs{ + sv: fullSv, + mhash: ComputeMembershipHash(bobOnly), + }) + + require.True(t, s.suppress) +} + +func TestEncodePartialSenderFirst(t *testing.T) { + tu.SetT(t) + + alice := tu.NoErr(enc.NameFromStr("/ndn/alice")) + bob := tu.NoErr(enc.NameFromStr("/ndn/bob")) + carol := tu.NoErr(enc.NameFromStr("/ndn/carol")) + + m := NewSvMap[uint64](0) + m.Set(alice.TlvStr(), 100, 5) + m.Set(bob.TlvStr(), 150, 3) + m.Set(carol.TlvStr(), 150, 7) + + // Threshold large enough for sender + one peer. + full := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + FullStateVector: &spec_svs.FullStateVector{StateVector: m.Encode(func(s uint64) uint64 { return s })}, + } + threshold := len(full.Encode().Join()) - 1 + + partial := encodePartialStateVector(m, PartialEncodeOpts{ + Sender: carol, + Threshold: threshold, + Mtime: map[string]time.Time{ + alice.TlvStr(): time.Unix(10, 0), + bob.TlvStr(): time.Unix(20, 0), + }, + }) + + require.NotEmpty(t, partial.Entries) + require.Equal(t, carol, partial.Entries[0].Name) + if len(partial.Entries) > 2 { + require.Less(t, partial.Entries[1].Name.Compare(partial.Entries[2].Name), 0) + } +} + +func TestBuildSvsDataForSendPublicationPartial(t *testing.T) { + tu.SetT(t) + + alice := tu.NoErr(enc.NameFromStr("/ndn/alice")) + m := NewSvMap[uint64](0) + m.Set(alice.TlvStr(), 100, 5) + for i := range 20 { + name := tu.NoErr(enc.NameFromStr(fmt.Sprintf("/ndn/peer%d", i))) + m.Set(name.TlvStr(), 150, uint64(i+1)) + } + + full := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + FullStateVector: &spec_svs.FullStateVector{StateVector: m.Encode(func(s uint64) uint64 { return s })}, + } + threshold := len(full.Encode().Join()) / 2 + + pub := buildSvsDataForSend(svsSendInput{ + State: m, Reason: syncSendPublication, Threshold: threshold, Sender: alice, + }) + require.True(t, pub.IsPartial()) + require.False(t, pub.IsFull()) + pubSv := pub.GetStateVector() + fullSv := full.GetStateVector() + require.Less(t, len(pubSv.Entries), len(fullSv.Entries)) + + periodic := buildSvsDataForSend(svsSendInput{ + State: m, Reason: syncSendPeriodic, Threshold: threshold, Sender: alice, + }) + require.True(t, periodic.IsFull()) + require.False(t, periodic.IsPartial()) +} + +func TestOnReceivePartialMergesPresentEntriesOnly(t *testing.T) { + tu.SetT(t) + + var updates []SvSyncUpdate + s := &SvSync{ + o: SvSyncOpts{ + OnUpdate: func(u SvSyncUpdate) { updates = append(updates, u) }, + PeriodicTimeout: 30 * time.Second, + }, + state: NewSvMap[uint64](0), + mtime: make(map[string]time.Time), + ticker: time.NewTicker(30 * time.Second), + } + + alice := tu.NoErr(enc.NameFromStr("/ndn/alice")) + bob := tu.NoErr(enc.NameFromStr("/ndn/bob")) + s.state.Set(alice.TlvStr(), 100, 1) + s.state.Set(bob.TlvStr(), 150, 1) + + bobOnly := NewSvMap[uint64](0) + bobOnly.Set(bob.TlvStr(), 150, 4) + partialSv := bobOnly.Encode(func(s uint64) uint64 { return s }) + + s.onReceiveStateVector(svSyncRecvSvArgs{ + sv: partialSv, + partial: true, + mhash: ComputeMembershipHash(bobOnly), + }) + + require.Len(t, updates, 1) + require.Equal(t, bob, updates[0].Name) + require.EqualValues(t, 4, updates[0].High) + require.EqualValues(t, 1, s.state.Get(alice.TlvStr(), 100)) + require.EqualValues(t, 4, s.state.Get(bob.TlvStr(), 150)) +} + +// --- mhash tests --- + +func TestComputeMembershipHashStable(t *testing.T) { + tu.SetT(t) + + m := NewSvMap[uint64](0) + m.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 1) + m.Set(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150, 3) + + h1 := ComputeMembershipHash(m) + h2 := ComputeMembershipHash(m) + require.Equal(t, h1, h2) + require.Len(t, h1, 32) +} + +func TestComputeMembershipHashOrderIndependent(t *testing.T) { + tu.SetT(t) + + m1 := NewSvMap[uint64](0) + m1.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 1) + m1.Set(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150, 3) + + m2 := NewSvMap[uint64](0) + m2.Set(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150, 3) + m2.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 1) + + require.Equal(t, ComputeMembershipHash(m1), ComputeMembershipHash(m2)) +} + +func TestComputeMembershipHashChangesOnMembership(t *testing.T) { + tu.SetT(t) + + m := NewSvMap[uint64](0) + m.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 1) + before := ComputeMembershipHash(m) + + m.Set(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150, 3) + after := ComputeMembershipHash(m) + require.NotEqual(t, before, after) +} + +func TestComputeMembershipHashIgnoresSeqNo(t *testing.T) { + tu.SetT(t) + + m1 := NewSvMap[uint64](0) + m1.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 1) + + m2 := NewSvMap[uint64](0) + m2.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 99) + + require.Equal(t, ComputeMembershipHash(m1), ComputeMembershipHash(m2)) +} + +func TestSvsDataInlineTLV(t *testing.T) { + tu.SetT(t) + + m := NewSvMap[uint64](0) + m.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 1) + sv := m.Encode(func(s uint64) uint64 { return s }) + + original := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + FullStateVector: &spec_svs.FullStateVector{StateVector: sv}, + } + wire := original.Encode().Join() + + parsed, err := spec_svs.ParseSvsData(enc.NewBufferView(wire), false) + require.NoError(t, err) + require.True(t, parsed.IsFull()) + require.False(t, parsed.IsPartial()) + require.Equal(t, original.MemberSetHash, parsed.MemberSetHash) + origSv := original.GetStateVector() + parsedSv := parsed.GetStateVector() + require.Equal(t, origSv.Entries[0].Name.String(), parsedSv.Entries[0].Name.String()) +} + +func TestSvsDataPublishTLV(t *testing.T) { + tu.SetT(t) + + ref := tu.NoErr(enc.NameFromStr("/ndn/svs/alice/100/32=sv/1")) + mhash := make([]byte, 32) + + original := &spec_svs.SvsData{MemberSetHash: mhash, SvsDataRef: ref} + wire := original.Encode().Join() + + parsed, err := spec_svs.ParseSvsData(enc.NewBufferView(wire), false) + require.NoError(t, err) + require.Equal(t, spec_svs.VectorKindNone, parsed.Kind()) + require.Equal(t, ref.String(), parsed.SvsDataRef.String()) + require.Nil(t, parsed.GetStateVector()) + require.Equal(t, mhash, parsed.MemberSetHash) +} + +func TestSvsDataPartialTLV(t *testing.T) { + tu.SetT(t) + + m := NewSvMap[uint64](0) + m.Set(tu.NoErr(enc.NameFromStr("/ndn/alice")).TlvStr(), 100, 1) + sv := m.Encode(func(s uint64) uint64 { return s }) + + original := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + PartialStateVector: &spec_svs.PartialStateVector{StateVector: sv}, + } + wire := original.Encode().Join() + + parsed, err := spec_svs.ParseSvsData(enc.NewBufferView(wire), false) + require.NoError(t, err) + require.True(t, parsed.IsPartial()) + require.False(t, parsed.IsFull()) + require.Equal(t, spec_svs.VectorKindPartial, parsed.Kind()) + require.NotNil(t, parsed.GetStateVector()) + require.Equal(t, ComputeMembershipHash(m), parsed.MemberSetHash) +} + +// --- pull / recovery tests --- + +func TestDeriveFullVectorPrefix(t *testing.T) { + tu.SetT(t) + + syncData := tu.NoErr(enc.NameFromStr("/ndn/svs/alice/1700000000/32=svs")) + prefix := deriveFullVectorPrefix(syncData) + require.Equal(t, "/ndn/svs/alice/1700000000/32=sv", prefix.String()) +} + +func TestPullRefFromSyncDataWire(t *testing.T) { + tu.SetT(t) + + syncDataName := tu.NoErr(enc.NameFromStr("/ndn/svs/alice/1700000000/32=svs")). + Append(enc.NewVersionComponent(12345)) + dataWire, err := spec.Spec{}.MakeData( + syncDataName, + &ndn.DataConfig{ContentType: optional.Some(ndn.ContentTypeBlob)}, + enc.Wire{enc.Buffer{0x01}}, + sig.NewSha256Signer(), + ) + require.NoError(t, err) + + ref := pullRefFromSyncDataWire(dataWire.Wire) + require.Equal(t, "/ndn/svs/alice/1700000000/32=sv", ref.String()) +} + +func TestBuildPublishSvsData(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + ref := tu.NoErr(enc.NameFromStr("/ndn/svs/alice/1700000000/32=sv/999")) + data := buildPublishSvsData(m, ref) + + require.Equal(t, spec_svs.VectorKindNone, data.Kind()) + require.Equal(t, ComputeMembershipHash(m), data.MemberSetHash) + require.True(t, ref.Equal(data.SvsDataRef)) + require.Nil(t, data.GetStateVector()) + + wire := data.Encode().Join() + parsed, err := spec_svs.ParseSvsData(enc.NewBufferView(wire), false) + require.NoError(t, err) + require.Equal(t, spec_svs.VectorKindNone, parsed.Kind()) + require.Equal(t, ComputeMembershipHash(m), parsed.MemberSetHash) + require.True(t, ref.Equal(parsed.SvsDataRef)) + require.Nil(t, parsed.GetStateVector()) +} + +func TestShouldUsePublishPull(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + sv := m.Encode(func(s uint64) uint64 { return s }) + fullSize := len((&spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + FullStateVector: &spec_svs.FullStateVector{StateVector: sv}, + }).Encode().Join()) + + usePublish, _, _ := shouldUsePublishPull(syncSendPublication, fullSize-1, m) + require.False(t, usePublish) + usePublish, _, _ = shouldUsePublishPull(syncSendPeriodic, fullSize+1, m) + require.False(t, usePublish) + usePublish, _, _ = shouldUsePublishPull(syncSendPeriodic, fullSize-1, m) + require.True(t, usePublish) + usePublish, _, _ = shouldUsePublishPull(syncSendOther, fullSize-1, m) + require.True(t, usePublish) + usePublish, _, _ = shouldUsePublishPull(syncSendRecovery, fullSize-1, m) + require.True(t, usePublish) +} + +func TestIsTrustedSvsDataRef(t *testing.T) { + tu.SetT(t) + + trust := tu.NoErr(enc.NameFromStr("/ndn/svs/alice/1/32=sv")) + ref := tu.NoErr(enc.NameFromStr("/ndn/svs/alice/1/32=sv/999")) + bad := tu.NoErr(enc.NameFromStr("/ndn/evil/32=sv/1")) + + require.True(t, isTrustedSvsDataRef(ref, trust)) + require.True(t, isTrustedSvsDataRef(trust, trust)) + require.False(t, isTrustedSvsDataRef(bad, trust)) + require.False(t, isTrustedSvsDataRef(ref, nil)) +} + +func TestParseFullVectorContentRejectsBadMhash(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + inline := &spec_svs.SvsData{ + MemberSetHash: []byte("not-a-valid-mhash-padding-000000"), + FullStateVector: &spec_svs.FullStateVector{StateVector: m.Encode(func(s uint64) uint64 { return s })}, + } + wire := inline.Encode().Join() + + _, err := parseFullVectorContent(wire) + require.Error(t, err) +} + +func TestParseFullVectorContentRejectsPartial(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + // Wrong wire TLV: PartialStateVector instead of FullStateVector. + inline := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + PartialStateVector: &spec_svs.PartialStateVector{StateVector: m.Encode(func(s uint64) uint64 { return s })}, + } + wire := inline.Encode().Join() + + _, err := parseFullVectorContent(wire) + require.Error(t, err) +} + +func TestParseFullVectorContentRejectsMissingMhash(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + inline := &spec_svs.SvsData{ + MemberSetHash: nil, + FullStateVector: &spec_svs.FullStateVector{StateVector: m.Encode(func(s uint64) uint64 { return s })}, + } + wire := inline.Encode().Join() + + _, err := parseFullVectorContent(wire) + require.Error(t, err) +} + +func TestParseFullVectorContent(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + inline := &spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + FullStateVector: &spec_svs.FullStateVector{StateVector: m.Encode(func(s uint64) uint64 { return s })}, + } + wire := inline.Encode().Join() + + parsed, err := parseFullVectorContent(wire) + require.NoError(t, err) + require.Equal(t, inline.MemberSetHash, parsed.MemberSetHash) + require.Len(t, parsed.GetStateVector().Entries, 2) +} + +func TestOnPulledFullVectorMergesState(t *testing.T) { + tu.SetT(t) + + var updates []SvSyncUpdate + s := &SvSync{ + o: SvSyncOpts{ + OnUpdate: func(u SvSyncUpdate) { updates = append(updates, u) }, + PeriodicTimeout: 30 * time.Second, + }, + state: NewSvMap[uint64](0), + mtime: make(map[string]time.Time), + ticker: time.NewTicker(30 * time.Second), + recvSv: make(chan svSyncRecvSvArgs, 1), + } + + alice := tu.NoErr(enc.NameFromStr("/ndn/alice")) + s.state.Set(alice.TlvStr(), 100, 1) + + remote := testSvMapAliceBob() + content := (&spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(remote), + FullStateVector: &spec_svs.FullStateVector{StateVector: remote.Encode(func(s uint64) uint64 { return s })}, + }).Encode().Join() + + go func() { + s.onPulledFullVector(content) + }() + + s.onReceiveStateVector(<-s.recvSv) + + require.Len(t, updates, 2) + require.EqualValues(t, 3, s.state.Get(tu.NoErr(enc.NameFromStr("/ndn/bob")).TlvStr(), 150)) + require.EqualValues(t, 5, s.state.Get(alice.TlvStr(), 100)) +} + +func TestEncodeSyncDataPublishMode(t *testing.T) { + tu.SetT(t) + + m := testSvMapAliceBob() + fullSize := len((&spec_svs.SvsData{ + MemberSetHash: ComputeMembershipHash(m), + FullStateVector: &spec_svs.FullStateVector{StateVector: m.Encode(func(s uint64) uint64 { return s })}, + }).Encode().Join()) + usePublish, _, _ := shouldUsePublishPull(syncSendPeriodic, fullSize-1, m) + require.True(t, usePublish) + + publish := buildPublishSvsData(m, tu.NoErr(enc.NameFromStr("/ndn/svs/alice/1/32=sv/2"))) + require.Equal(t, spec_svs.VectorKindNone, publish.Kind()) + require.Nil(t, publish.GetStateVector()) +}