add flush_dht_record bindings

veilid-flutter/example/integration_test/app_test.dart

@@ -118,6 +118,7 @@ void main() {
               tags: ['dht']);
           test('inspect dht record', testDHT.testInspectDHTRecord,
               tags: ['dht']);
+          test('flush dht record', testDHT.testFlushDHTRecord, tags: ['dht']);
         });
 
         group('DHT Transactions $cryptoKind', () {

veilid-flutter/example/integration_test/test_dht.dart

@@ -296,4 +296,26 @@ class TestDHT {
     await routingContext.closeDHTRecord(rec.key);
     await routingContext.deleteDHTRecord(rec.key);
   }
+
+  Future<void> testFlushDHTRecord() async {
+    final rec = await routingContext.createDHTRecord(
+        cryptoKind, const DHTSchema.dflt(oCnt: 1));
+
+    expect(
+        await routingContext.setDHTValue(
+            rec.key, 0, utf8.encode('flush test value')),
+        isNull);
+
+    // flush with no timeout should wait until writes drain and return true
+    expect(await routingContext.flushDHTRecord(rec.key), isTrue);
+
+    // no pending writes — tight timeout should still return true immediately
+    expect(
+        await routingContext.flushDHTRecord(rec.key,
+            timeout: const Duration(milliseconds: 10)),
+        isTrue);
+
+    await routingContext.closeDHTRecord(rec.key);
+    await routingContext.deleteDHTRecord(rec.key);
+  }
 }

veilid-flutter/lib/routing_context.dart

@@ -57,4 +57,5 @@ abstract class VeilidRoutingContext {
     List<ValueSubkeyRange>? subkeys,
     DHTReportScope scope = DHTReportScope.local,
   });
+  Future<bool> flushDHTRecord(RecordKey key, {Duration? timeout});
 }

veilid-flutter/lib/veilid_ffi.dart

@@ -101,6 +101,10 @@ typedef _RoutingContextCancelDHTWatchDart =
 //     id: u32, key: FfiStr, subkeys: FfiStr, scope: FfiStr)
 typedef _RoutingContextInspectDHTRecordDart =
     void Function(int, int, Pointer<Utf8>, Pointer<Utf8>, Pointer<Utf8>);
+// fn routing_context_flush_dht_record(port: i64,
+//     id: u32, key: FfiStr, timeout_ms: u64)
+typedef _RoutingContextFlushDHTRecordDart =
+    void Function(int, int, Pointer<Utf8>, int);
 
 // fn generate_member_id(port: i64, writer_key: FfiStr)
 typedef _GenerateMemberIdDart = void Function(int, Pointer<Utf8>);
@@ -957,6 +961,22 @@ class VeilidRoutingContextFFI extends VeilidRoutingContext {
     );
     return report;
   }
+
+  @override
+  Future<bool> flushDHTRecord(RecordKey key, {Duration? timeout}) async {
+    _ctx.ensureValid();
+    final nativeKey = jsonEncode(key).toNativeUtf8();
+
+    final recvPort = ReceivePort('routing_context_flush_dht_record');
+    final sendPort = recvPort.sendPort;
+    _ctx.ffi._routingContextFlushDHTRecord(
+      sendPort.nativePort,
+      _ctx.id!,
+      nativeKey,
+      timeout?.inMilliseconds ?? 0,
+    );
+    return processFuturePlain<bool>(recvPort.first);
+  }
 }
 
 class _TDBT {
@@ -1739,6 +1759,7 @@ class VeilidFFI extends Veilid {
   final _RoutingContextWatchDHTValuesDart _routingContextWatchDHTValues;
   final _RoutingContextCancelDHTWatchDart _routingContextCancelDHTWatch;
   final _RoutingContextInspectDHTRecordDart _routingContextInspectDHTRecord;
+  final _RoutingContextFlushDHTRecordDart _routingContextFlushDHTRecord;
 
   final _GenerateMemberIdDart _generateMemberId;
   final _GetDHTRecordKeyDart _getDHTRecordKey;
@@ -1968,6 +1989,11 @@ class VeilidFFI extends Veilid {
             ),
             _RoutingContextInspectDHTRecordDart
           >('routing_context_inspect_dht_record'),
+      _routingContextFlushDHTRecord = dylib
+          .lookupFunction<
+            Void Function(Int64, Uint32, Pointer<Utf8>, Uint64),
+            _RoutingContextFlushDHTRecordDart
+          >('routing_context_flush_dht_record'),
       _generateMemberId = dylib
           .lookupFunction<
             Void Function(Int64, Pointer<Utf8>),

veilid-flutter/lib/veilid_js.dart

@@ -390,6 +390,16 @@ class VeilidRoutingContextJS extends VeilidRoutingContext {
       ],
     );
   }
+
+  @override
+  Future<bool> flushDHTRecord(RecordKey key, {Duration? timeout}) {
+    final id = _ctx.requireId();
+    return _wasmPromiseBool('routing_context_flush_dht_record', [
+      id.toJS,
+      jsonEncode(key).toJS,
+      if (timeout != null) timeout.inMilliseconds.toJS else null,
+    ]);
+  }
 }
 
 // JS implementation of VeilidCryptoSystem

veilid-flutter/rust/src/dart_ffi.rs

@@ -904,6 +904,32 @@ pub extern "C" fn routing_context_inspect_dht_record(
     );
 }
 
+#[no_mangle]
+#[instrument(level = "trace", target = "ffi", skip_all)]
+pub extern "C" fn routing_context_flush_dht_record(
+    port: i64,
+    id: u32,
+    key: FfiStr,
+    timeout_ms: u64,
+) {
+    let key: RecordKey = deserialize_opt_json(key.into_opt_string()).unwrap_or_log();
+    let timeout = if timeout_ms == 0 {
+        None
+    } else {
+        Some(Duration::from_millis(timeout_ms))
+    };
+
+    DartIsolateWrapper::new(port).spawn_result(
+        async move {
+            let routing_context = get_routing_context(id, "routing_context_flush_dht_record")?;
+
+            let res = routing_context.flush_dht_record(key, timeout).await?;
+            VeilidAPIResult::Ok(res)
+        }
+        .in_current_span(),
+    );
+}
+
 #[no_mangle]
 #[instrument(level = "trace", target = "ffi", skip_all)]
 pub extern "C" fn generate_member_id(port: i64, writer_key: FfiStr) {

veilid-python/tests/test_dht.py

@@ -877,6 +877,25 @@ async def test_schema_limit_smpl(api_connection: veilid.VeilidAPI):
 
 
 
+@pytest.mark.asyncio
+async def test_flush_dht_record(api_connection: veilid.VeilidAPI):
+    rc = await api_connection.new_routing_context()
+    async with rc:
+        for kind in await api_connection.valid_crypto_kinds():
+            rec = await rc.create_dht_record(kind, veilid.DHTSchema.dflt(1))
+
+            await rc.set_dht_value(rec.key, ValueSubkey(0), b"flush test value")
+
+            # flush with no timeout should wait until writes drain and return true
+            assert await rc.flush_dht_record(rec.key) is True
+
+            # no pending writes — tight timeout should still return true immediately
+            assert await rc.flush_dht_record(rec.key, 10) is True
+
+            await rc.close_dht_record(rec.key)
+            await rc.delete_dht_record(rec.key)
+
+
 @pytest.mark.skipif(os.getenv("INTEGRATION") != "1", reason="integration test requires two servers running")
 @pytest.mark.asyncio
 async def test_dht_integration_writer_reader():

veilid-python/veilid/api.py

@@ -115,6 +115,14 @@ class RoutingContext(ABC):
     ) -> types.DHTRecordReport:
         pass
 
+    @abstractmethod
+    async def flush_dht_record(
+        self, 
+        key: types.RecordKey, 
+        timeout_ms: Optional[int] = None
+    ) -> bool:
+        pass
+
 
 
 class TableDbTransaction(ABC):

veilid-python/veilid/json_api.py

@@ -866,6 +866,25 @@ class _JsonRoutingContext(RoutingContext):
             )
         )
 
+    async def flush_dht_record(
+        self, 
+        key: RecordKey, 
+        timeout_ms: Optional[int] = None
+    ) -> bool:
+        assert isinstance(key, RecordKey)
+        assert timeout_ms is None or isinstance(timeout_ms, int)
+
+        return raise_api_result(
+            await self.api.send_ndjson_request(
+                Operation.ROUTING_CONTEXT,
+                validate=validate_rc_op,
+                rc_id=self.rc_id,
+                rc_op=RoutingContextOperation.FLUSH_DHT_RECORD,
+                key=key,
+                timeout_ms=timeout_ms,
+            )
+        )
+
 
 
 

veilid-python/veilid/operations.py

@@ -55,6 +55,7 @@ class RoutingContextOperation(StrEnum):
     WATCH_DHT_VALUES = "WatchDhtValues"
     CANCEL_DHT_WATCH = "CancelDhtWatch"
     INSPECT_DHT_RECORD = "InspectDhtRecord"
+    FLUSH_DHT_RECORD = "FlushDhtRecord"
 
 
 class TableDbOperation(StrEnum):

veilid-python/veilid/schema/RecvMessage.json

@@ -3300,6 +3300,42 @@
           "required": [
             "rc_op"
           ]
+        },
+        {
+          "type": "object",
+          "properties": {
+            "rc_op": {
+              "type": "string",
+              "const": "FlushDhtRecord"
+            }
+          },
+          "anyOf": [
+            {
+              "type": "object",
+              "properties": {
+                "value": {
+                  "type": "boolean"
+                }
+              },
+              "required": [
+                "value"
+              ]
+            },
+            {
+              "type": "object",
+              "properties": {
+                "error": {
+                  "$ref": "#/$defs/VeilidAPIError"
+                }
+              },
+              "required": [
+                "error"
+              ]
+            }
+          ],
+          "required": [
+            "rc_op"
+          ]
         }
       ],
       "required": [

veilid-python/veilid/schema/Request.json

@@ -1792,6 +1792,30 @@
             "rc_op",
             "key"
           ]
+        },
+        {
+          "type": "object",
+          "properties": {
+            "key": {
+              "type": "string"
+            },
+            "rc_op": {
+              "type": "string",
+              "const": "FlushDhtRecord"
+            },
+            "timeout_ms": {
+              "type": [
+                "integer",
+                "null"
+              ],
+              "format": "uint64",
+              "minimum": 0
+            }
+          },
+          "required": [
+            "rc_op",
+            "key"
+          ]
         }
       ],
       "required": [

veilid-remote-api/src/process.rs

@@ -386,6 +386,15 @@ impl JsonRequestProcessor {
                         .map(Box::new),
                 ),
             },
+            RoutingContextRequestOp::FlushDhtRecord { key, timeout_ms } => {
+                RoutingContextResponseOp::FlushDhtRecord {
+                    result: to_json_api_result(
+                        routing_context
+                            .flush_dht_record(key, timeout_ms.map(Duration::from_millis))
+                            .await,
+                    ),
+                }
+            }
         };
         RoutingContextResponse {
             rc_id: rcr.rc_id,

veilid-remote-api/src/routing_context.rs

@@ -93,6 +93,11 @@ pub enum RoutingContextRequestOp {
         #[schemars(default)]
         scope: DHTReportScope,
     },
+    FlushDhtRecord {
+        #[schemars(with = "String")]
+        key: RecordKey,
+        timeout_ms: Option<u64>,
+    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
@@ -159,4 +164,8 @@ pub enum RoutingContextResponseOp {
         #[serde(flatten)]
         result: ApiResult<Box<DHTRecordReport>>,
     },
+    FlushDhtRecord {
+        #[serde(flatten)]
+        result: ApiResult<bool>,
+    },
 }

veilid-wasm/src/dart/mod.rs

@@ -582,6 +582,21 @@ pub fn routing_context_inspect_dht_record(
     })
 }
 
+#[wasm_bindgen()]
+pub fn routing_context_flush_dht_record(id: u32, key: String, timeout_ms: Option<u32>) -> Promise {
+    wrap_api_future_plain(async move {
+        let key: veilid_core::RecordKey =
+            veilid_core::deserialize_json(&key).map_err(VeilidAPIError::generic)?;
+
+        let routing_context = get_routing_context(id, "routing_context_flush_dht_record")?;
+
+        let res = routing_context
+            .flush_dht_record(key, timeout_ms.map(|ms| Duration::from_millis(ms as u64)))
+            .await?;
+        VeilidAPIResult::Ok(res)
+    })
+}
+
 #[wasm_bindgen()]
 pub fn generate_member_id(writer_key: String) -> Promise {
     wrap_api_future_json(async move {

veilid-wasm/src/js/veilid_routing_context_js.rs

@@ -360,4 +360,21 @@ impl VeilidRoutingContext {
             .inspect_dht_record(recordKey.clone(), subkeys, scope)
             .await
     }
+
+    /// Waits for any pending offline subkey writes for a DHT record to be flushed to the network.
+    /// `timeoutMs` is the maximum time to wait in milliseconds, or `undefined` to wait indefinitely.
+    /// Returns `true` if all pending writes were flushed, `false` if the timeout elapsed first.
+    pub async fn flushDHTRecord(
+        &self,
+        recordKey: &RecordKey,
+        timeoutMs: Option<u32>,
+    ) -> VeilidAPIResult<bool> {
+        let routing_context = self.getRoutingContext()?;
+        routing_context
+            .flush_dht_record(
+                recordKey.clone(),
+                timeoutMs.map(|ms| Duration::from_millis(ms as u64)),
+            )
+            .await
+    }
 }

veilid-wasm/tests/src/VeilidRoutingContext.test.ts

@@ -439,6 +439,23 @@ describe('VeilidRoutingContext', () => {
           await expect(inspectRes.localSeqs).toEqual([0]);
           await expect(inspectRes.networkSeqs).toEqual([undefined]);
         });
+
+        it('should flush a record and confirm no offline subkeys remain', async () => {
+          const setValueRes = await routingContext.setDHTValue(
+            dhtRecord.key,
+            0,
+            textEncoder.encode(data)
+          );
+          await expect(setValueRes).toBeUndefined();
+
+          // flush with no timeout should wait until writes drain and return true
+          const flushRes = await routingContext.flushDHTRecord(dhtRecord.key);
+          await expect(flushRes).toBe(true);
+
+          // no pending writes — tight timeout should still return true immediately
+          const flushResWithTimeout = await routingContext.flushDHTRecord(dhtRecord.key, 10);
+          await expect(flushResWithTimeout).toBe(true);
+        });
       });
 
       describe(`DHT transactions simple tests for ${cryptoKind as string}`, () => {