short array work

This commit is contained in:
Christien Rioux 2023-07-30 01:11:24 -04:00
parent bb8a3df281
commit 672bcd5c3d
7 changed files with 543 additions and 104 deletions

View File

@ -536,15 +536,15 @@ class DHTData extends $pb.GeneratedMessage {
void clearSize() => clearField(4);
}
class DHTList extends $pb.GeneratedMessage {
factory DHTList() => create();
DHTList._() : super();
factory DHTList.fromBuffer($core.List<$core.int> i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromBuffer(i, r);
factory DHTList.fromJson($core.String i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromJson(i, r);
class DHTShortArray extends $pb.GeneratedMessage {
factory DHTShortArray() => create();
DHTShortArray._() : super();
factory DHTShortArray.fromBuffer($core.List<$core.int> i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromBuffer(i, r);
factory DHTShortArray.fromJson($core.String i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromJson(i, r);
static final $pb.BuilderInfo _i = $pb.BuilderInfo(_omitMessageNames ? '' : 'DHTList', createEmptyInstance: create)
static final $pb.BuilderInfo _i = $pb.BuilderInfo(_omitMessageNames ? '' : 'DHTShortArray', createEmptyInstance: create)
..pc<TypedKey>(1, _omitFieldNames ? '' : 'keys', $pb.PbFieldType.PM, subBuilder: TypedKey.create)
..p<$core.int>(2, _omitFieldNames ? '' : 'index', $pb.PbFieldType.KU3)
..a<$core.List<$core.int>>(2, _omitFieldNames ? '' : 'index', $pb.PbFieldType.OY)
..hasRequiredFields = false
;
@ -552,28 +552,34 @@ class DHTList extends $pb.GeneratedMessage {
'Using this can add significant overhead to your binary. '
'Use [GeneratedMessageGenericExtensions.deepCopy] instead. '
'Will be removed in next major version')
DHTList clone() => DHTList()..mergeFromMessage(this);
DHTShortArray clone() => DHTShortArray()..mergeFromMessage(this);
@$core.Deprecated(
'Using this can add significant overhead to your binary. '
'Use [GeneratedMessageGenericExtensions.rebuild] instead. '
'Will be removed in next major version')
DHTList copyWith(void Function(DHTList) updates) => super.copyWith((message) => updates(message as DHTList)) as DHTList;
DHTShortArray copyWith(void Function(DHTShortArray) updates) => super.copyWith((message) => updates(message as DHTShortArray)) as DHTShortArray;
$pb.BuilderInfo get info_ => _i;
@$core.pragma('dart2js:noInline')
static DHTList create() => DHTList._();
DHTList createEmptyInstance() => create();
static $pb.PbList<DHTList> createRepeated() => $pb.PbList<DHTList>();
static DHTShortArray create() => DHTShortArray._();
DHTShortArray createEmptyInstance() => create();
static $pb.PbList<DHTShortArray> createRepeated() => $pb.PbList<DHTShortArray>();
@$core.pragma('dart2js:noInline')
static DHTList getDefault() => _defaultInstance ??= $pb.GeneratedMessage.$_defaultFor<DHTList>(create);
static DHTList? _defaultInstance;
static DHTShortArray getDefault() => _defaultInstance ??= $pb.GeneratedMessage.$_defaultFor<DHTShortArray>(create);
static DHTShortArray? _defaultInstance;
@$pb.TagNumber(1)
$core.List<TypedKey> get keys => $_getList(0);
@$pb.TagNumber(2)
$core.List<$core.int> get index => $_getList(1);
$core.List<$core.int> get index => $_getN(1);
@$pb.TagNumber(2)
set index($core.List<$core.int> v) { $_setBytes(1, v); }
@$pb.TagNumber(2)
$core.bool hasIndex() => $_has(1);
@$pb.TagNumber(2)
void clearIndex() => clearField(2);
}
class DHTLog extends $pb.GeneratedMessage {
@ -1111,6 +1117,74 @@ class Profile extends $pb.GeneratedMessage {
TypedKey ensureAvatar() => $_ensure(4);
}
class OwnedDHTRecordPointer extends $pb.GeneratedMessage {
factory OwnedDHTRecordPointer() => create();
OwnedDHTRecordPointer._() : super();
factory OwnedDHTRecordPointer.fromBuffer($core.List<$core.int> i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromBuffer(i, r);
factory OwnedDHTRecordPointer.fromJson($core.String i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromJson(i, r);
static final $pb.BuilderInfo _i = $pb.BuilderInfo(_omitMessageNames ? '' : 'OwnedDHTRecordPointer', createEmptyInstance: create)
..aOM<TypedKey>(1, _omitFieldNames ? '' : 'recordKey', subBuilder: TypedKey.create)
..aOM<CryptoKey>(2, _omitFieldNames ? '' : 'ownerKey', subBuilder: CryptoKey.create)
..aOM<CryptoKey>(3, _omitFieldNames ? '' : 'ownerSecret', subBuilder: CryptoKey.create)
..hasRequiredFields = false
;
@$core.Deprecated(
'Using this can add significant overhead to your binary. '
'Use [GeneratedMessageGenericExtensions.deepCopy] instead. '
'Will be removed in next major version')
OwnedDHTRecordPointer clone() => OwnedDHTRecordPointer()..mergeFromMessage(this);
@$core.Deprecated(
'Using this can add significant overhead to your binary. '
'Use [GeneratedMessageGenericExtensions.rebuild] instead. '
'Will be removed in next major version')
OwnedDHTRecordPointer copyWith(void Function(OwnedDHTRecordPointer) updates) => super.copyWith((message) => updates(message as OwnedDHTRecordPointer)) as OwnedDHTRecordPointer;
$pb.BuilderInfo get info_ => _i;
@$core.pragma('dart2js:noInline')
static OwnedDHTRecordPointer create() => OwnedDHTRecordPointer._();
OwnedDHTRecordPointer createEmptyInstance() => create();
static $pb.PbList<OwnedDHTRecordPointer> createRepeated() => $pb.PbList<OwnedDHTRecordPointer>();
@$core.pragma('dart2js:noInline')
static OwnedDHTRecordPointer getDefault() => _defaultInstance ??= $pb.GeneratedMessage.$_defaultFor<OwnedDHTRecordPointer>(create);
static OwnedDHTRecordPointer? _defaultInstance;
@$pb.TagNumber(1)
TypedKey get recordKey => $_getN(0);
@$pb.TagNumber(1)
set recordKey(TypedKey v) { setField(1, v); }
@$pb.TagNumber(1)
$core.bool hasRecordKey() => $_has(0);
@$pb.TagNumber(1)
void clearRecordKey() => clearField(1);
@$pb.TagNumber(1)
TypedKey ensureRecordKey() => $_ensure(0);
@$pb.TagNumber(2)
CryptoKey get ownerKey => $_getN(1);
@$pb.TagNumber(2)
set ownerKey(CryptoKey v) { setField(2, v); }
@$pb.TagNumber(2)
$core.bool hasOwnerKey() => $_has(1);
@$pb.TagNumber(2)
void clearOwnerKey() => clearField(2);
@$pb.TagNumber(2)
CryptoKey ensureOwnerKey() => $_ensure(1);
@$pb.TagNumber(3)
CryptoKey get ownerSecret => $_getN(2);
@$pb.TagNumber(3)
set ownerSecret(CryptoKey v) { setField(3, v); }
@$pb.TagNumber(3)
$core.bool hasOwnerSecret() => $_has(2);
@$pb.TagNumber(3)
void clearOwnerSecret() => clearField(3);
@$pb.TagNumber(3)
CryptoKey ensureOwnerSecret() => $_ensure(2);
}
class Account extends $pb.GeneratedMessage {
factory Account() => create();
Account._() : super();
@ -1121,8 +1195,8 @@ class Account extends $pb.GeneratedMessage {
..aOM<Profile>(1, _omitFieldNames ? '' : 'profile', subBuilder: Profile.create)
..aOB(2, _omitFieldNames ? '' : 'invisible')
..a<$core.int>(3, _omitFieldNames ? '' : 'autoAwayTimeoutSec', $pb.PbFieldType.OU3)
..aOM<TypedKey>(4, _omitFieldNames ? '' : 'contactList', subBuilder: TypedKey.create)
..aOM<TypedKey>(5, _omitFieldNames ? '' : 'contactRequests', subBuilder: TypedKey.create)
..aOM<OwnedDHTRecordPointer>(4, _omitFieldNames ? '' : 'contactList', subBuilder: OwnedDHTRecordPointer.create)
..aOM<OwnedDHTRecordPointer>(5, _omitFieldNames ? '' : 'contactRequests', subBuilder: OwnedDHTRecordPointer.create)
..hasRequiredFields = false
;
@ -1177,26 +1251,26 @@ class Account extends $pb.GeneratedMessage {
void clearAutoAwayTimeoutSec() => clearField(3);
@$pb.TagNumber(4)
TypedKey get contactList => $_getN(3);
OwnedDHTRecordPointer get contactList => $_getN(3);
@$pb.TagNumber(4)
set contactList(TypedKey v) { setField(4, v); }
set contactList(OwnedDHTRecordPointer v) { setField(4, v); }
@$pb.TagNumber(4)
$core.bool hasContactList() => $_has(3);
@$pb.TagNumber(4)
void clearContactList() => clearField(4);
@$pb.TagNumber(4)
TypedKey ensureContactList() => $_ensure(3);
OwnedDHTRecordPointer ensureContactList() => $_ensure(3);
@$pb.TagNumber(5)
TypedKey get contactRequests => $_getN(4);
OwnedDHTRecordPointer get contactRequests => $_getN(4);
@$pb.TagNumber(5)
set contactRequests(TypedKey v) { setField(5, v); }
set contactRequests(OwnedDHTRecordPointer v) { setField(5, v); }
@$pb.TagNumber(5)
$core.bool hasContactRequests() => $_has(4);
@$pb.TagNumber(5)
void clearContactRequests() => clearField(5);
@$pb.TagNumber(5)
TypedKey ensureContactRequests() => $_ensure(4);
OwnedDHTRecordPointer ensureContactRequests() => $_ensure(4);
}
class ContactInvitation extends $pb.GeneratedMessage {

View File

@ -163,19 +163,19 @@ final $typed_data.Uint8List dHTDataDescriptor = $convert.base64Decode(
'kuVHlwZWRLZXlSBGhhc2gSFAoFY2h1bmsYAyABKA1SBWNodW5rEhIKBHNpemUYBCABKA1SBHNp'
'emU=');
@$core.Deprecated('Use dHTListDescriptor instead')
const DHTList$json = {
'1': 'DHTList',
@$core.Deprecated('Use dHTShortArrayDescriptor instead')
const DHTShortArray$json = {
'1': 'DHTShortArray',
'2': [
{'1': 'keys', '3': 1, '4': 3, '5': 11, '6': '.TypedKey', '10': 'keys'},
{'1': 'index', '3': 2, '4': 3, '5': 13, '10': 'index'},
{'1': 'index', '3': 2, '4': 1, '5': 12, '10': 'index'},
],
};
/// Descriptor for `DHTList`. Decode as a `google.protobuf.DescriptorProto`.
final $typed_data.Uint8List dHTListDescriptor = $convert.base64Decode(
'CgdESFRMaXN0Eh0KBGtleXMYASADKAsyCS5UeXBlZEtleVIEa2V5cxIUCgVpbmRleBgCIAMoDV'
'IFaW5kZXg=');
/// Descriptor for `DHTShortArray`. Decode as a `google.protobuf.DescriptorProto`.
final $typed_data.Uint8List dHTShortArrayDescriptor = $convert.base64Decode(
'Cg1ESFRTaG9ydEFycmF5Eh0KBGtleXMYASADKAsyCS5UeXBlZEtleVIEa2V5cxIUCgVpbmRleB'
'gCIAEoDFIFaW5kZXg=');
@$core.Deprecated('Use dHTLogDescriptor instead')
const DHTLog$json = {
@ -308,6 +308,22 @@ final $typed_data.Uint8List profileDescriptor = $convert.base64Decode(
'eVIMYXZhaWxhYmlsaXR5EiYKBmF2YXRhchgFIAEoCzIJLlR5cGVkS2V5SABSBmF2YXRhcogBAU'
'IJCgdfYXZhdGFy');
@$core.Deprecated('Use ownedDHTRecordPointerDescriptor instead')
const OwnedDHTRecordPointer$json = {
'1': 'OwnedDHTRecordPointer',
'2': [
{'1': 'record_key', '3': 1, '4': 1, '5': 11, '6': '.TypedKey', '10': 'recordKey'},
{'1': 'owner_key', '3': 2, '4': 1, '5': 11, '6': '.CryptoKey', '10': 'ownerKey'},
{'1': 'owner_secret', '3': 3, '4': 1, '5': 11, '6': '.CryptoKey', '10': 'ownerSecret'},
],
};
/// Descriptor for `OwnedDHTRecordPointer`. Decode as a `google.protobuf.DescriptorProto`.
final $typed_data.Uint8List ownedDHTRecordPointerDescriptor = $convert.base64Decode(
'ChVPd25lZERIVFJlY29yZFBvaW50ZXISKAoKcmVjb3JkX2tleRgBIAEoCzIJLlR5cGVkS2V5Ug'
'lyZWNvcmRLZXkSJwoJb3duZXJfa2V5GAIgASgLMgouQ3J5cHRvS2V5Ughvd25lcktleRItCgxv'
'd25lcl9zZWNyZXQYAyABKAsyCi5DcnlwdG9LZXlSC293bmVyU2VjcmV0');
@$core.Deprecated('Use accountDescriptor instead')
const Account$json = {
'1': 'Account',
@ -315,8 +331,8 @@ const Account$json = {
{'1': 'profile', '3': 1, '4': 1, '5': 11, '6': '.Profile', '10': 'profile'},
{'1': 'invisible', '3': 2, '4': 1, '5': 8, '10': 'invisible'},
{'1': 'auto_away_timeout_sec', '3': 3, '4': 1, '5': 13, '10': 'autoAwayTimeoutSec'},
{'1': 'contact_list', '3': 4, '4': 1, '5': 11, '6': '.TypedKey', '10': 'contactList'},
{'1': 'contact_requests', '3': 5, '4': 1, '5': 11, '6': '.TypedKey', '10': 'contactRequests'},
{'1': 'contact_list', '3': 4, '4': 1, '5': 11, '6': '.OwnedDHTRecordPointer', '10': 'contactList'},
{'1': 'contact_requests', '3': 5, '4': 1, '5': 11, '6': '.OwnedDHTRecordPointer', '10': 'contactRequests'},
],
};
@ -324,9 +340,9 @@ const Account$json = {
final $typed_data.Uint8List accountDescriptor = $convert.base64Decode(
'CgdBY2NvdW50EiIKB3Byb2ZpbGUYASABKAsyCC5Qcm9maWxlUgdwcm9maWxlEhwKCWludmlzaW'
'JsZRgCIAEoCFIJaW52aXNpYmxlEjEKFWF1dG9fYXdheV90aW1lb3V0X3NlYxgDIAEoDVISYXV0'
'b0F3YXlUaW1lb3V0U2VjEiwKDGNvbnRhY3RfbGlzdBgEIAEoCzIJLlR5cGVkS2V5Ugtjb250YW'
'N0TGlzdBI0ChBjb250YWN0X3JlcXVlc3RzGAUgASgLMgkuVHlwZWRLZXlSD2NvbnRhY3RSZXF1'
'ZXN0cw==');
'b0F3YXlUaW1lb3V0U2VjEjkKDGNvbnRhY3RfbGlzdBgEIAEoCzIWLk93bmVkREhUUmVjb3JkUG'
'9pbnRlclILY29udGFjdExpc3QSQQoQY29udGFjdF9yZXF1ZXN0cxgFIAEoCzIWLk93bmVkREhU'
'UmVjb3JkUG9pbnRlclIPY29udGFjdFJlcXVlc3Rz');
@$core.Deprecated('Use contactInvitationDescriptor instead')
const ContactInvitation$json = {

View File

@ -71,7 +71,7 @@ message DHTData {
uint32 size = 4;
}
// DHTList - represents a re-orderable collection of individual elements
// DHTShortArray - represents a re-orderable collection of up to 256 individual elements
// Header in subkey 0 of first key follows this structure
//
// stride = descriptor subkey count on first key - 1
@ -79,16 +79,17 @@ message DHTData {
// Subkeys 0..stride on the 'keys' keys are also individual elements
//
// Keys must use writable schema in order to make this list mutable
message DHTList {
message DHTShortArray {
// Other keys to concatenate
// Uses the same writer as this DHTList with SMPL schema
repeated TypedKey keys = 1;
// Item position index
// Item position index (uint8[256])
// Actual item location is:
// idx = index[n] + 1 (offset for header at idx 0)
// key = idx / stride
// subkey = idx % stride
repeated uint32 index = 2;
bytes index = 2;
// Free items are not represented in the list but can be
// calculated through iteration
}
@ -250,7 +251,7 @@ message Account {
// The contacts DHTList for this account
// DHT Private
OwnedDHTRecordPointer contact_list = 4;
// The contact requests DHTList for this account
// The contact requests DHTShortArray for this account
// DHT Private
OwnedDHTRecordPointer contact_requests = 5;
}

View File

@ -9,14 +9,15 @@ import 'package:veilid/veilid.dart';
import '../entities/entities.dart';
import '../entities/proto.dart' as proto;
import '../tools/tools.dart';
import '../veilid_support/dht_short_array.dart';
import '../veilid_support/veilid_support.dart';
import 'logins.dart';
part 'contact_request_records.g.dart';
// Contact invitation records stored in Account
class ContactRequestRecords extends DHTList<proto.ContactRequestRecord> {
//
class ContactRequestRecords {
DHTShortArray _backingArray;
Future<proto.ContactRequestRecord> newContactRequest(
proto.EncryptionKind encryptionKind,

View File

@ -1,43 +0,0 @@
import 'dart:typed_data';
import 'package:protobuf/protobuf.dart';
import 'package:veilid/veilid.dart';
import '../tools/tools.dart';
import 'veilid_support.dart';
class DHTList {
DHTList({required DHTRecord dhtRecord}) : _dhtRecord = dhtRecord;
final DHTRecord _dhtRecord;
static Future<DHTList> create(VeilidRoutingContext dhtctx,
{DHTRecordCrypto? crypto}) async {
final dhtRecord = await DHTRecord.create(dhtctx, crypto: crypto);
final dhtList = DHTList(dhtRecord: dhtRecord);
return dhtList;
}
static Future<DHTList> openRead(
VeilidRoutingContext dhtctx, TypedKey dhtRecordKey,
{DHTRecordCrypto? crypto}) async {
final dhtRecord =
await DHTRecord.openRead(dhtctx, dhtRecordKey, crypto: crypto);
final dhtList = DHTList(dhtRecord: dhtRecord);
return dhtList;
}
static Future<DHTList> openWrite(
VeilidRoutingContext dhtctx,
TypedKey dhtRecordKey,
KeyPair writer, {
DHTRecordCrypto? crypto,
}) async {
final dhtRecord =
await DHTRecord.openWrite(dhtctx, dhtRecordKey, writer, crypto: crypto);
final dhtList = DHTList(dhtRecord: dhtRecord);
return dhtList;
}
////////////////////////////////////////////////////////////////
}

View File

@ -1,9 +1,9 @@
import 'dart:async';
import 'dart:typed_data';
import 'package:protobuf/protobuf.dart';
import 'package:veilid/veilid.dart';
import '../entities/proto.dart' as proto;
import '../tools/tools.dart';
import 'veilid_support.dart';
@ -17,11 +17,13 @@ class DHTRecord {
: _dhtctx = dhtctx,
_recordDescriptor = recordDescriptor,
_defaultSubkey = defaultSubkey,
_writer = writer;
_writer = writer,
_subkeySeqCache = {};
final VeilidRoutingContext _dhtctx;
final DHTRecordDescriptor _recordDescriptor;
final int _defaultSubkey;
final KeyPair? _writer;
final Map<int, int> _subkeySeqCache;
DHTRecordCrypto crypto;
static Future<DHTRecord> create(VeilidRoutingContext dhtctx,
@ -76,12 +78,11 @@ class DHTRecord {
int subkeyOrDefault(int subkey) => (subkey == -1) ? _defaultSubkey : subkey;
VeilidRoutingContext get routingContext => _dhtctx;
TypedKey get key => _recordDescriptor.key;
PublicKey get owner => _recordDescriptor.owner;
KeyPair? get ownerKeyPair => _recordDescriptor.ownerKeyPair();
DHTSchema get schema => _recordDescriptor.schema;
KeyPair? get writer => _writer;
Future<void> close() async {
@ -111,19 +112,31 @@ class DHTRecord {
}
}
Future<Uint8List?> get({int subkey = -1, bool forceRefresh = false}) async {
Future<Uint8List?> get(
{int subkey = -1,
bool forceRefresh = false,
bool onlyUpdates = false}) async {
subkey = subkeyOrDefault(subkey);
final valueData =
await _dhtctx.getDHTValue(_recordDescriptor.key, subkey, false);
await _dhtctx.getDHTValue(_recordDescriptor.key, subkey, forceRefresh);
if (valueData == null) {
return null;
}
return crypto.decrypt(valueData.data, subkey);
final lastSeq = _subkeySeqCache[subkey];
if (lastSeq != null && valueData.seq <= lastSeq) {
return null;
}
final out = crypto.decrypt(valueData.data, subkey);
_subkeySeqCache[subkey] = valueData.seq;
return out;
}
Future<T?> getJson<T>(T Function(dynamic) fromJson,
{int subkey = -1, bool forceRefresh = false}) async {
final data = await get(subkey: subkey, forceRefresh: forceRefresh);
{int subkey = -1,
bool forceRefresh = false,
bool onlyUpdates = false}) async {
final data = await get(
subkey: subkey, forceRefresh: forceRefresh, onlyUpdates: onlyUpdates);
if (data == null) {
return null;
}
@ -133,18 +146,34 @@ class DHTRecord {
Future<T?> getProtobuf<T extends GeneratedMessage>(
T Function(List<int> i) fromBuffer,
{int subkey = -1,
bool forceRefresh = false}) async {
final data = await get(subkey: subkey, forceRefresh: forceRefresh);
bool forceRefresh = false,
bool onlyUpdates = false}) async {
final data = await get(
subkey: subkey, forceRefresh: forceRefresh, onlyUpdates: onlyUpdates);
if (data == null) {
return null;
}
return fromBuffer(data.toList());
}
Future<Uint8List?> tryWriteBytes(Uint8List newValue,
{int subkey = -1}) async {
subkey = subkeyOrDefault(subkey);
newValue = await crypto.encrypt(newValue, subkey);
// Set the new data if possible
final valueData =
await _dhtctx.setDHTValue(_recordDescriptor.key, subkey, newValue);
if (valueData == null) {
return null;
}
return valueData.data;
}
Future<void> eventualWriteBytes(Uint8List newValue, {int subkey = -1}) async {
subkey = subkeyOrDefault(subkey);
newValue = await crypto.encrypt(newValue, subkey);
// Get existing identity key
ValueData? valueData;
do {
// Set the new data
@ -159,14 +188,17 @@ class DHTRecord {
Future<Uint8List> Function(Uint8List oldValue) update,
{int subkey = -1}) async {
subkey = subkeyOrDefault(subkey);
// Get existing identity key
// Get existing identity key, do not allow force refresh here
// because if we need a refresh the setDHTValue will fail anyway
var valueData =
await _dhtctx.getDHTValue(_recordDescriptor.key, subkey, false);
// Ensure it exists already
if (valueData == null) {
throw const FormatException('value does not exist');
}
do {
// Ensure it exists already
if (valueData == null) {
throw const FormatException('value does not exist');
}
// Update cache
_subkeySeqCache[subkey] = valueData!.seq;
// Update the data
final oldData = await crypto.decrypt(valueData.data, subkey);
@ -181,6 +213,25 @@ class DHTRecord {
} while (valueData != null);
}
Future<T?> tryWriteJson<T>(T Function(dynamic) fromJson, T newValue,
{int subkey = -1}) =>
tryWriteBytes(jsonEncodeBytes(newValue), subkey: subkey).then((out) {
if (out == null) {
return null;
}
return jsonDecodeBytes(fromJson, out);
});
Future<T?> tryWriteProtobuf<T extends GeneratedMessage>(
T Function(List<int>) fromBuffer, T newValue,
{int subkey = -1}) =>
tryWriteBytes(newValue.writeToBuffer(), subkey: subkey).then((out) {
if (out == null) {
return null;
}
return fromBuffer(out);
});
Future<void> eventualWriteJson<T>(T newValue, {int subkey = -1}) =>
eventualWriteBytes(jsonEncodeBytes(newValue), subkey: subkey);

View File

@ -0,0 +1,339 @@
import 'dart:typed_data';
import 'package:protobuf/protobuf.dart';
import 'package:veilid/veilid.dart';
import '../entities/proto.dart' as proto;
import '../tools/tools.dart';
import 'veilid_support.dart';
class _DHTShortArrayCache {
_DHTShortArrayCache()
: linkedRecords = List<DHTRecord>.empty(growable: true),
index = List<int>.empty(growable: true),
free = List<int>.empty(growable: true);
final List<DHTRecord> linkedRecords;
final List<int> index;
final List<int> free;
}
class DHTShortArray {
DHTShortArray({required DHTRecord dhtRecord})
: _headRecord = dhtRecord,
_head = _DHTShortArrayCache() {
late final int stride;
switch (dhtRecord.schema) {
case DHTSchemaDFLT(oCnt: final oCnt):
stride = oCnt - 1;
if (stride <= 0) {
throw StateError('Invalid stride in DHTShortArray');
}
case DHTSchemaSMPL():
throw StateError('Wrote kind of DHT record for DHTShortArray');
}
_stride = stride;
}
// Head DHT record
final DHTRecord _headRecord;
late final int _stride;
// Cached representation refreshed from head record
_DHTShortArrayCache _head;
static Future<DHTShortArray> create(VeilidRoutingContext dhtctx, int stride,
{DHTRecordCrypto? crypto}) async {
final dhtRecord = await DHTRecord.create(dhtctx,
schema: DHTSchema.dflt(oCnt: stride + 1), crypto: crypto);
final dhtShortArray = DHTShortArray(dhtRecord: dhtRecord);
return dhtShortArray;
}
static Future<DHTShortArray> openRead(
VeilidRoutingContext dhtctx, TypedKey dhtRecordKey,
{DHTRecordCrypto? crypto}) async {
final dhtRecord =
await DHTRecord.openRead(dhtctx, dhtRecordKey, crypto: crypto);
final dhtShortArray = DHTShortArray(dhtRecord: dhtRecord);
await dhtShortArray._refreshHead();
return dhtShortArray;
}
static Future<DHTShortArray> openWrite(
VeilidRoutingContext dhtctx,
TypedKey dhtRecordKey,
KeyPair writer, {
DHTRecordCrypto? crypto,
}) async {
final dhtRecord =
await DHTRecord.openWrite(dhtctx, dhtRecordKey, writer, crypto: crypto);
final dhtShortArray = DHTShortArray(dhtRecord: dhtRecord);
await dhtShortArray._refreshHead();
return dhtShortArray;
}
////////////////////////////////////////////////////////////////
/// Write the current head cache out to a protobuf to be serialized
Uint8List _headToBuffer() {
final head = proto.DHTShortArray();
head.keys.addAll(_head.linkedRecords.map((lr) => lr.key.toProto()));
head.index.addAll(_head.index);
return head.writeToBuffer();
}
Future<DHTRecord> _openLinkedRecord(TypedKey recordKey) async {
final writer = _headRecord.writer;
return (writer != null)
? await DHTRecord.openWrite(
_headRecord.routingContext, recordKey, writer)
: await DHTRecord.openRead(_headRecord.routingContext, recordKey);
}
/// Validate the head from the DHT is properly formatted
/// and calculate the free list from it while we're here
List<int> _validateHeadCacheData(
List<Typed<FixedEncodedString43>> linkedKeys, List<int> index) {
// Ensure nothing is duplicated in the linked keys set
final newKeys = linkedKeys.toSet();
assert(newKeys.length == linkedKeys.length, 'duplicated linked keys');
final newIndex = index.toSet();
assert(newIndex.length == newIndex.length, 'duplicated index locations');
// Ensure all the index keys fit into the existing records
final indexCount = (linkedKeys.length + 1) * _stride;
int? maxIndex;
for (final idx in newIndex) {
assert(idx >= 0 || idx < indexCount, 'index out of range');
if (maxIndex == null || idx > maxIndex) {
maxIndex = idx;
}
}
final free = <int>[];
if (maxIndex != null) {
for (var i = 0; i < maxIndex; i++) {
if (!newIndex.contains(i)) {
free.add(i);
}
}
}
return free;
}
Future<bool> _refreshHead(
{bool forceRefresh = false, 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);
if (head == null) {
if (onlyUpdates) {
// No update
return false;
}
throw StateError('head missing during initial refresh');
}
// Get the set of new linked keys and validate it
final linkedKeys = head.keys.map(proto.TypedKeyProto.fromProto).toList();
final index = head.index;
final free = _validateHeadCacheData(linkedKeys, index);
// See which records are actually new
final oldRecords = Map<TypedKey, DHTRecord>.fromEntries(
_head.linkedRecords.map((lr) => MapEntry(lr.key, lr)));
final newRecords = <TypedKey, DHTRecord>{};
final sameRecords = <TypedKey, DHTRecord>{};
try {
for (var n = 0; n < linkedKeys.length; n++) {
final newKey = linkedKeys[n];
final oldRecord = oldRecords[newKey];
if (oldRecord == null) {
// Open the new record
final newRecord = await _openLinkedRecord(newKey);
newRecords[newKey] = newRecord;
} else {
sameRecords[newKey] = oldRecord;
}
}
} on Exception catch (_) {
// On any exception close the records we have opened
await Future.wait(newRecords.entries.map((e) => e.value.close()));
}
// From this point forward we should not throw an exception or everything
// is possibly invalid. Just pass the exception up it happens and the caller
// will have to delete this short array and reopen it if it can
await Future.wait(oldRecords.entries
.where((e) => !sameRecords.containsKey(e.key))
.map((e) => e.value.close()));
// Figure out which indices are free
// Make the new head cache
_head = _DHTShortArrayCache()
..linkedRecords.addAll(
linkedKeys.map((key) => (sameRecords[key] ?? newRecords[key])!))
..index.addAll(index)
..free.addAll(free);
return true;
}
////////////////////////////////////////////////////////////////
Future<void> close() async {
final futures = <Future<void>>[_headRecord.close()];
for (final lr in _head.linkedRecords) {
futures.add(lr.close());
}
await Future.wait(futures);
}
Future<void> delete() async {
final futures = <Future<void>>[_headRecord.close()];
for (final lr in _head.linkedRecords) {
futures.add(lr.delete());
}
await Future.wait(futures);
}
Future<T> scope<T>(Future<T> Function(DHTShortArray) scopeFunction) async {
try {
return await scopeFunction(this);
} finally {
await close();
}
}
Future<T> deleteScope<T>(
Future<T> Function(DHTShortArray) scopeFunction) async {
try {
final out = await scopeFunction(this);
await close();
return out;
} on Exception catch (_) {
await delete();
rethrow;
}
}
DHTRecord? _getRecord(int recordNumber) {
if (recordNumber == 0) {
return _headRecord;
}
recordNumber--;
if (recordNumber >= _head.linkedRecords.length) {
return null;
}
return _head.linkedRecords[recordNumber];
}
// xxx: add
// xxx: insert
// xxx: swap
// xxx: remove
// xxx: clear
Future<Uint8List?> getItem(int index, {bool forceRefresh = false}) async {
await _refreshHead(forceRefresh: forceRefresh, onlyUpdates: true);
if (index < 0 || index >= _head.index.length) {
throw IndexError.withLength(index, _head.index.length);
}
final recordNumber = index ~/ _stride;
final record = _getRecord(recordNumber);
assert(record != null, 'Record does not exist');
final recordSubkey = (index % _stride) + ((recordNumber == 0) ? 1 : 0);
return record!.get(subkey: recordSubkey, forceRefresh: forceRefresh);
}
Future<Uint8List?> tryWriteItem(int index, Uint8List newValue) async {
if (await _refreshHead(onlyUpdates: true)) {
throw StateError('structure changed');
}
if (index < 0 || index >= _head.index.length) {
throw IndexError.withLength(index, _head.index.length);
}
final recordNumber = index ~/ _stride;
final record = _getRecord(recordNumber);
assert(record != null, 'Record does not exist');
final recordSubkey = (index % _stride) + ((recordNumber == 0) ? 1 : 0);
return record!.tryWriteBytes(newValue, subkey: recordSubkey);
}
Future<void> eventualWriteItem(int index, Uint8List newValue) async {
Uint8List? oldData;
do {
// Set it back
oldData = await tryWriteItem(index, newValue);
// Repeat if newer data on the network was found
} while (oldData != null);
}
Future<void> eventualUpdateItem(
int index, Future<Uint8List> Function(Uint8List oldValue) update) async {
var oldData = await getItem(index);
// Ensure it exists already
if (oldData == null) {
throw const FormatException('value does not exist');
}
do {
// Update the data
final updatedData = await update(oldData!);
// Set it back
oldData = await tryWriteItem(index, updatedData);
// Repeat if newer data on the network was found
} while (oldData != null);
}
Future<T?> tryWriteItemJson<T>(
T Function(dynamic) fromJson,
int index,
T newValue,
) =>
tryWriteItem(index, jsonEncodeBytes(newValue)).then((out) {
if (out == null) {
return null;
}
return jsonDecodeBytes(fromJson, out);
});
Future<T?> tryWriteItemProtobuf<T extends GeneratedMessage>(
T Function(List<int>) fromBuffer,
int index,
T newValue,
) =>
tryWriteItem(index, newValue.writeToBuffer()).then((out) {
if (out == null) {
return null;
}
return fromBuffer(out);
});
Future<void> eventualWriteItemJson<T>(int index, T newValue) =>
eventualWriteItem(index, jsonEncodeBytes(newValue));
Future<void> eventualWriteItemProtobuf<T extends GeneratedMessage>(
int index, T newValue,
{int subkey = -1}) =>
eventualWriteItem(index, newValue.writeToBuffer());
Future<void> eventualUpdateItemJson<T>(
T Function(dynamic) fromJson,
int index,
Future<T> Function(T) update,
) =>
eventualUpdateItem(index, jsonUpdate(fromJson, update));
Future<void> eventualUpdateItemProtobuf<T extends GeneratedMessage>(
T Function(List<int>) fromBuffer,
int index,
Future<T> Function(T) update,
) =>
eventualUpdateItem(index, protobufUpdate(fromBuffer, update));
}