watch for dhtshortarray

packages/veilid_support/lib/dht_support/src/dht_record.dart

@@ -1,6 +1,8 @@
 import 'dart:async';
 import 'dart:typed_data';
 
+import 'package:fast_immutable_collections/fast_immutable_collections.dart';
+import 'package:meta/meta.dart';
 import 'package:protobuf/protobuf.dart';
 
 import '../../../veilid_support.dart';
@@ -31,9 +33,13 @@ class DHTRecord {
   final DHTRecordCrypto _crypto;
   bool _open;
   bool _valid;
+  @internal
   StreamController<VeilidUpdateValueChange>? watchController;
+  @internal
   bool needsWatchStateUpdate;
+  @internal
   bool inWatchStateUpdate;
+  @internal
   WatchState? watchState;
 
   int subkeyOrDefault(int subkey) => (subkey == -1) ? _defaultSubkey : subkey;
@@ -45,6 +51,7 @@ class DHTRecord {
   DHTSchema get schema => _recordDescriptor.schema;
   int get subkeyCount => _recordDescriptor.schema.subkeyCount();
   KeyPair? get writer => _writer;
+  DHTRecordCrypto get crypto => _crypto;
   OwnedDHTRecordPointer get ownedDHTRecordPointer =>
       OwnedDHTRecordPointer(recordKey: key, owner: ownerKeyPair!);
 
@@ -266,9 +273,12 @@ class DHTRecord {
       Timestamp? expiration,
       int? count}) async {
     // Set up watch requirements which will get picked up by the next tick
-    watchState =
-        WatchState(subkeys: subkeys, expiration: expiration, count: count);
-    needsWatchStateUpdate = true;
+    final oldWatchState = watchState;
+    watchState = WatchState(
+        subkeys: subkeys?.lock, expiration: expiration, count: count);
+    if (oldWatchState != watchState) {
+      needsWatchStateUpdate = true;
+    }
   }
 
   Future<StreamSubscription<VeilidUpdateValueChange>> listen(
@@ -294,7 +304,9 @@ class DHTRecord {
 
   Future<void> cancelWatch() async {
     // Tear down watch requirements
-    watchState = null;
-    needsWatchStateUpdate = true;
+    if (watchState != null) {
+      watchState = null;
+      needsWatchStateUpdate = true;
+    }
   }
 }

packages/veilid_support/lib/dht_support/src/dht_record_pool.dart

@@ -1,5 +1,6 @@
 import 'dart:async';
 
+import 'package:equatable/equatable.dart';
 import 'package:fast_immutable_collections/fast_immutable_collections.dart';
 import 'package:freezed_annotation/freezed_annotation.dart';
 
@@ -37,13 +38,19 @@ class OwnedDHTRecordPointer with _$OwnedDHTRecordPointer {
 }
 
 /// Watch state
-class WatchState {
-  WatchState(
-      {required this.subkeys, required this.expiration, required this.count});
-  List<ValueSubkeyRange>? subkeys;
-  Timestamp? expiration;
-  int? count;
-  Timestamp? realExpiration;
+class WatchState extends Equatable {
+  const WatchState(
+      {required this.subkeys,
+      required this.expiration,
+      required this.count,
+      this.realExpiration});
+  final IList<ValueSubkeyRange>? subkeys;
+  final Timestamp? expiration;
+  final int? count;
+  final Timestamp? realExpiration;
+
+  @override
+  List<Object?> get props => [subkeys, expiration, count, realExpiration];
 }
 
 class DHTRecordPool with TableDBBacked<DHTRecordPoolAllocations> {
@@ -400,11 +407,17 @@ class DHTRecordPool with TableDBBacked<DHTRecordPoolAllocations> {
               try {
                 final realExpiration = await kv.value.routingContext
                     .watchDHTValues(kv.key,
-                        subkeys: ws.subkeys,
+                        subkeys: ws.subkeys?.toList(),
                         count: ws.count,
                         expiration: ws.expiration);
                 kv.value.needsWatchStateUpdate = false;
-                ws.realExpiration = realExpiration;
+
+                // Update watch state with real expiration
+                kv.value.watchState = WatchState(
+                    subkeys: ws.subkeys,
+                    expiration: ws.expiration,
+                    count: ws.count,
+                    realExpiration: realExpiration);
               } on VeilidAPIException {
                 // Failed to cancel DHT watch, try again next tick
               }

packages/veilid_support/lib/dht_support/src/dht_short_array.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 import 'dart:typed_data';
 
+import 'package:mutex/mutex.dart';
 import 'package:protobuf/protobuf.dart';
 
 import '../../../../veilid_support.dart';
@@ -32,7 +33,9 @@ class _DHTShortArrayCache {
 class DHTShortArray {
   DHTShortArray._({required DHTRecord headRecord})
       : _headRecord = headRecord,
-        _head = _DHTShortArrayCache() {
+        _head = _DHTShortArrayCache(),
+        _subscriptions = {},
+        _listenMutex = Mutex() {
     late final int stride;
     switch (headRecord.schema) {
       case DHTSchemaDFLT(oCnt: final oCnt):
@@ -59,6 +62,14 @@ class DHTShortArray {
   // Cached representation refreshed from head record
   _DHTShortArrayCache _head;
 
+  // Subscription to head and linked record internal changes
+  final Map<TypedKey, StreamSubscription<VeilidUpdateValueChange>>
+      _subscriptions;
+  // Stream of external changes
+  StreamController<void>? _watchController;
+  // Watch mutex to ensure we keep the representation valid
+  final Mutex _listenMutex;
+
   // Create a DHTShortArray
   // if smplWriter is specified, uses a SMPL schema with a single writer
   // rather than the key owner
@@ -273,6 +284,11 @@ class DHTShortArray {
           linkedKeys.map((key) => (sameRecords[key] ?? newRecords[key])!))
       ..index.addAll(index)
       ..free.addAll(free);
+
+    // Update watch if we have one in case linked records have been added
+    if (_watchController != null) {
+      await _watchAllRecords();
+    }
   }
 
   /// Pull the latest or updated copy of the head record from the network
@@ -280,7 +296,7 @@ class DHTShortArray {
       {bool forceRefresh = true, bool onlyUpdates = false}) async {
     // Get an updated head record copy if one exists
     final head = await _headRecord.getProtobuf(proto.DHTShortArray.fromBuffer,
-        forceRefresh: forceRefresh, onlyUpdates: onlyUpdates);
+        subkey: 0, forceRefresh: forceRefresh, onlyUpdates: onlyUpdates);
     if (head == null) {
       if (onlyUpdates) {
         // No update
@@ -297,6 +313,7 @@ class DHTShortArray {
   ////////////////////////////////////////////////////////////////
 
   Future<void> close() async {
+    await _watchController?.close();
     final futures = <Future<void>>[_headRecord.close()];
     for (final lr in _head.linkedRecords) {
       futures.add(lr.close());
@@ -305,7 +322,9 @@ class DHTShortArray {
   }
 
   Future<void> delete() async {
-    final futures = <Future<void>>[_headRecord.close()];
+    await _watchController?.close();
+
+    final futures = <Future<void>>[_headRecord.delete()];
     for (final lr in _head.linkedRecords) {
       futures.add(lr.delete());
     }
@@ -332,7 +351,7 @@ class DHTShortArray {
     }
   }
 
-  DHTRecord? _getRecord(int recordNumber) {
+  DHTRecord? _getLinkedRecord(int recordNumber) {
     if (recordNumber == 0) {
       return _headRecord;
     }
@@ -343,6 +362,43 @@ class DHTShortArray {
     return _head.linkedRecords[recordNumber];
   }
 
+  Future<DHTRecord> _getOrCreateLinkedRecord(int recordNumber) async {
+    if (recordNumber == 0) {
+      return _headRecord;
+    }
+    final pool = DHTRecordPool.instance;
+    recordNumber--;
+    while (recordNumber >= _head.linkedRecords.length) {
+      // Linked records must use SMPL schema so writer can be specified
+      // Use the same writer as the head record
+      final smplWriter = _headRecord.writer!;
+      final parent = pool.getParentRecordKey(_headRecord.key);
+      final routingContext = _headRecord.routingContext;
+      final crypto = _headRecord.crypto;
+
+      final schema = DHTSchema.smpl(
+          oCnt: 0,
+          members: [DHTSchemaMember(mKey: smplWriter.key, mCnt: _stride)]);
+      final dhtCreateRecord = await pool.create(
+          parent: parent,
+          routingContext: routingContext,
+          schema: schema,
+          crypto: crypto,
+          writer: smplWriter);
+      // Reopen with SMPL writer
+      await dhtCreateRecord.close();
+      final dhtRecord = await pool.openWrite(dhtCreateRecord.key, smplWriter,
+          parent: parent, routingContext: routingContext, crypto: crypto);
+
+      // Add to linked records
+      _head.linkedRecords.add(dhtRecord);
+      if (!await _tryWriteHead()) {
+        await _refreshHead();
+      }
+    }
+    return _head.linkedRecords[recordNumber];
+  }
+
   int _emptyIndex() {
     if (_head.free.isNotEmpty) {
       return _head.free.removeLast();
@@ -368,7 +424,7 @@ class DHTShortArray {
     }
     final index = _head.index[pos];
     final recordNumber = index ~/ _stride;
-    final record = _getRecord(recordNumber);
+    final record = _getLinkedRecord(recordNumber);
     assert(record != null, 'Record does not exist');
 
     final recordSubkey = (index % _stride) + ((recordNumber == 0) ? 1 : 0);
@@ -472,7 +528,7 @@ class DHTShortArray {
       final removedIdx = _head.index.removeAt(pos);
       _freeIndex(removedIdx);
       final recordNumber = removedIdx ~/ _stride;
-      final record = _getRecord(recordNumber);
+      final record = _getLinkedRecord(recordNumber);
       assert(record != null, 'Record does not exist');
       final recordSubkey =
           (removedIdx % _stride) + ((recordNumber == 0) ? 1 : 0);
@@ -532,11 +588,10 @@ class DHTShortArray {
     final index = _head.index[pos];
 
     final recordNumber = index ~/ _stride;
-    final record = _getRecord(recordNumber);
-    assert(record != null, 'Record does not exist');
+    final record = await _getOrCreateLinkedRecord(recordNumber);
 
     final recordSubkey = (index % _stride) + ((recordNumber == 0) ? 1 : 0);
-    return record!.tryWriteBytes(newValue, subkey: recordSubkey);
+    return record.tryWriteBytes(newValue, subkey: recordSubkey);
   }
 
   Future<void> eventualWriteItem(int pos, Uint8List newValue) async {
@@ -609,32 +664,97 @@ class DHTShortArray {
   ) =>
       eventualUpdateItem(pos, protobufUpdate(fromBuffer, update));
 
-  Future<void> watch() async {
-    // Watch head and all linked records
+  // Watch head and all linked records
+  Future<void> _watchAllRecords() async {
+    // This will update any existing watches if necessary
     try {
       await [_headRecord.watch(), ..._head.linkedRecords.map((r) => r.watch())]
           .wait;
-    } finally {
-      await [
-        _headRecord.cancelWatch(),
-        ..._head.linkedRecords.map((r) => r.cancelWatch())
-      ].wait;
+
+      // Update changes to the head record
+      if (!_subscriptions.containsKey(_headRecord.key)) {
+        _subscriptions[_headRecord.key] =
+            await _headRecord.listen(_onUpdateRecord);
+      }
+      // Update changes to any linked records
+      for (final lr in _head.linkedRecords) {
+        if (!_subscriptions.containsKey(lr.key)) {
+          _subscriptions[lr.key] = await lr.listen(_onUpdateRecord);
+        }
+      }
+    } on Exception {
+      // If anything fails, try to cancel the watches
+      await _cancelRecordWatches();
+      rethrow;
     }
   }
 
-  Future<void> listen(
-    Future<void> Function() onChanged,
-  ) async {
-    _headRecord.listen((update) => {
-      xxx
-    }
-  }
-
-  Future<void> cancelWatch() async {
-    // Watch head and all linked records
+  // Stop watching for changes to head and linked records
+  Future<void> _cancelRecordWatches() async {
     await _headRecord.cancelWatch();
     for (final lr in _head.linkedRecords) {
       await lr.cancelWatch();
     }
+    await _subscriptions.values.map((s) => s.cancel()).wait;
+    _subscriptions.clear();
   }
+
+  // Called when a head or linked record changes
+  Future<void> _onUpdateRecord(VeilidUpdateValueChange update) async {
+    final record = _head.linkedRecords.firstWhere(
+        (element) => element.key == update.key,
+        orElse: () => _headRecord);
+
+    // If head record subkey zero changes, then the layout
+    // of the dhtshortarray has changed
+    var updateHead = false;
+    if (record == _headRecord && update.subkeys.containsSubkey(0)) {
+      updateHead = true;
+    }
+
+    // If we have any other subkeys to update, do them first
+    final unord = List<Future<Uint8List?>>.empty(growable: true);
+    for (final skr in update.subkeys) {
+      for (var subkey = skr.low; subkey <= skr.high; subkey++) {
+        // Skip head subkey
+        if (subkey == 0) {
+          continue;
+        }
+        // Get the subkey, which caches the result in the local record store
+        unord.add(record.get(subkey: subkey, forceRefresh: true));
+      }
+    }
+    await unord.wait;
+
+    // Then update the head record
+    if (updateHead) {
+      await _refreshHead(forceRefresh: false);
+    }
+    // Then commit the change to any listeners
+    _watchController?.sink.add(null);
+  }
+
+  Future<StreamSubscription<void>> listen(
+    void Function() onChanged,
+  ) =>
+      _listenMutex.protect(() async {
+        // If don't have a controller yet, set it up
+        if (_watchController == null) {
+          // Set up watch requirements
+          _watchController = StreamController<void>.broadcast(onCancel: () {
+            // If there are no more listeners then we can get
+            // rid of the controller and drop our subscriptions
+            unawaited(_listenMutex.protect(() async {
+              // Cancel watches of head and linked records
+              await _cancelRecordWatches();
+              _watchController = null;
+            }));
+          });
+
+          // Start watching head and linked records
+          await _watchAllRecords();
+        }
+        // Return subscription
+        return _watchController!.stream.listen((_) => onChanged());
+      });
 }

packages/veilid_support/lib/dht_support/src/dht_short_array_cubit.dart

@@ -21,7 +21,7 @@ class DHTShortArrayCubit<T> extends Cubit<AsyncValue<IList<T>>> {
       } on Exception catch (e) {
         emit(AsyncValue.error(e));
       }
-
+xxx do this now
       shortArray. xxx add listen to head and linked records in dht_short_array
 
       _subscription = await record.listen((update) async {

packages/veilid_support/pubspec.lock

@@ -194,7 +194,7 @@ packages:
     source: hosted
     version: "2.3.4"
   equatable:
-    dependency: transitive
+    dependency: "direct main"
     description:
       name: equatable
       sha256: c2b87cb7756efdf69892005af546c56c0b5037f54d2a88269b4f347a505e3ca2
@@ -388,7 +388,7 @@ packages:
     source: hosted
     version: "0.5.0"
   meta:
-    dependency: transitive
+    dependency: "direct main"
     description:
       name: meta
       sha256: a6e590c838b18133bb482a2745ad77c5bb7715fb0451209e1a7567d416678b8e

packages/veilid_support/pubspec.yaml

@@ -8,10 +8,12 @@ environment:
 
 dependencies:
   bloc: ^8.1.2
+  equatable: ^2.0.5
   fast_immutable_collections: ^9.1.5
   freezed_annotation: ^2.2.0
   json_annotation: ^4.8.1
   loggy: ^2.0.3
+  meta: ^1.10.0
   mutex: ^3.1.0
   protobuf: ^3.0.0
   veilid: