mirror of
https://gitlab.com/veilid/veilidchat.git
synced 2025-08-09 06:32:27 -04:00
message integrity
This commit is contained in:
Christien Rioux
parent
490051a650
commit
fd63a0d5e0
11 changed files with 370 additions and 237 deletions
|
|
@ -1,145 +1,191 @@
|
|||
import 'dart:async';
|
||||
import 'dart:collection';
|
||||
import 'dart:math';
|
||||
|
||||
import 'package:fast_immutable_collections/fast_immutable_collections.dart';
|
||||
import 'package:veilid_support/veilid_support.dart';
|
||||
|
||||
import '../../../proto/proto.dart' as proto;
|
||||
|
||||
import 'author_input_source.dart';
|
||||
import 'message_integrity.dart';
|
||||
import 'output_position.dart';
|
||||
|
||||
class AuthorInputQueue {
|
||||
AuthorInputQueue({
|
||||
required this.author,
|
||||
required this.inputSource,
|
||||
required this.lastOutputPosition,
|
||||
required this.onError,
|
||||
}):
|
||||
assert(inputSource.messages.count>0, 'no input source window length'),
|
||||
assert(inputSource.messages.elements.isNotEmpty, 'no input source elements'),
|
||||
assert(inputSource.messages.tail >= inputSource.messages.elements.length, 'tail is before initial messages end'),
|
||||
assert(inputSource.messages.tail > 0, 'tail is not greater than zero'),
|
||||
currentPosition = inputSource.messages.tail,
|
||||
currentWindow = inputSource.messages.elements,
|
||||
windowLength = inputSource.messages.count,
|
||||
windowFirst = inputSource.messages.tail - inputSource.messages.elements.length,
|
||||
windowLast = inputSource.messages.tail - 1;
|
||||
AuthorInputQueue._({
|
||||
required TypedKey author,
|
||||
required AuthorInputSource inputSource,
|
||||
required OutputPosition? outputPosition,
|
||||
required void Function(Object, StackTrace?) onError,
|
||||
required MessageIntegrity messageIntegrity,
|
||||
}) : _author = author,
|
||||
_onError = onError,
|
||||
_inputSource = inputSource,
|
||||
_outputPosition = outputPosition,
|
||||
_lastMessage = outputPosition?.message,
|
||||
_messageIntegrity = messageIntegrity,
|
||||
_currentPosition = inputSource.currentWindow.last;
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
bool get isEmpty => toReconcile.isEmpty;
|
||||
|
||||
proto.Message? get current => toReconcile.firstOrNull;
|
||||
|
||||
bool consume() {
|
||||
toReconcile.removeFirst();
|
||||
return toReconcile.isNotEmpty;
|
||||
static Future<AuthorInputQueue?> create({
|
||||
required TypedKey author,
|
||||
required AuthorInputSource inputSource,
|
||||
required OutputPosition? outputPosition,
|
||||
required void Function(Object, StackTrace?) onError,
|
||||
}) async {
|
||||
final queue = AuthorInputQueue._(
|
||||
author: author,
|
||||
inputSource: inputSource,
|
||||
outputPosition: outputPosition,
|
||||
onError: onError,
|
||||
messageIntegrity: await MessageIntegrity.create(author: author));
|
||||
if (!await queue._findStartOfWork()) {
|
||||
return null;
|
||||
}
|
||||
return queue;
|
||||
}
|
||||
|
||||
Future<bool> prepareInputQueue() async {
|
||||
// Go through batches of the input dhtlog starting with
|
||||
// the current cubit state which is at the tail of the log
|
||||
// Find the last reconciled message for this author
|
||||
////////////////////////////////////////////////////////////////////////////
|
||||
// Public interface
|
||||
|
||||
outer:
|
||||
// Check if there are no messages in this queue to reconcile
|
||||
bool get isEmpty => _currentMessage == null;
|
||||
|
||||
// Get the current message that needs reconciliation
|
||||
proto.Message? get current => _currentMessage;
|
||||
|
||||
// Get the earliest output position to start inserting
|
||||
OutputPosition? get outputPosition => _outputPosition;
|
||||
|
||||
// Get the author of this queue
|
||||
TypedKey get author => _author;
|
||||
|
||||
// Remove a reconciled message and move to the next message
|
||||
// Returns true if there is more work to do
|
||||
Future<bool> consume() async {
|
||||
while (true) {
|
||||
for (var rn = currentWindow.length;
|
||||
rn >= 0 && currentPosition >= 0;
|
||||
rn--, currentPosition--) {
|
||||
final elem = currentWindow[rn];
|
||||
_lastMessage = _currentMessage;
|
||||
|
||||
// If we've found an input element that is older than our last
|
||||
// reconciled message for this author, then we stop
|
||||
if (lastOutputPosition != null) {
|
||||
if (elem.value.timestamp < lastOutputPosition!.message.timestamp) {
|
||||
break outer;
|
||||
}
|
||||
}
|
||||
_currentPosition++;
|
||||
|
||||
// Drop the 'offline' elements because we don't reconcile
|
||||
// anything until it has been confirmed to be committed to the DHT
|
||||
if (elem.isOffline) {
|
||||
// Get more window if we need to
|
||||
if (!await _updateWindow()) {
|
||||
// Window is not available so this queue can't work right now
|
||||
return false;
|
||||
}
|
||||
final nextMessage = _inputSource.currentWindow
|
||||
.elements[_currentPosition - _inputSource.currentWindow.first];
|
||||
|
||||
// Drop the 'offline' elements because we don't reconcile
|
||||
// anything until it has been confirmed to be committed to the DHT
|
||||
if (nextMessage.isOffline) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (_lastMessage != null) {
|
||||
// Ensure the timestamp is not moving backward
|
||||
if (nextMessage.value.timestamp < _lastMessage!.timestamp) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Add to head of reconciliation queue
|
||||
toReconcile.addFirst(elem.value);
|
||||
if (toReconcile.length > _maxQueueChunk) {
|
||||
toReconcile.removeLast();
|
||||
}
|
||||
}
|
||||
if (currentPosition < 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
xxx update window here and make this and other methods work
|
||||
// Verify the id chain for the message
|
||||
final matchId = await _messageIntegrity.generateMessageId(_lastMessage);
|
||||
if (matchId.compare(nextMessage.value.idBytes) != 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Verify the signature for the message
|
||||
if (!await _messageIntegrity.verifyMessage(nextMessage.value)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
_currentMessage = nextMessage.value;
|
||||
break;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////
|
||||
// Internal implementation
|
||||
|
||||
// Walk backward from the tail of the input queue to find the first
|
||||
// message newer than our last reconcicled message from this author
|
||||
// Returns false if no work is needed
|
||||
Future<bool> _findStartOfWork() async {
|
||||
// Iterate windows over the inputSource
|
||||
outer:
|
||||
while (true) {
|
||||
// Iterate through current window backward
|
||||
for (var i = _inputSource.currentWindow.elements.length;
|
||||
i >= 0 && _currentPosition >= 0;
|
||||
i--, _currentPosition--) {
|
||||
final elem = _inputSource.currentWindow.elements[i];
|
||||
|
||||
// If we've found an input element that is older than our last
|
||||
// reconciled message for this author, then we stop
|
||||
if (_lastMessage != null) {
|
||||
if (elem.value.timestamp < _lastMessage!.timestamp) {
|
||||
break outer;
|
||||
}
|
||||
}
|
||||
}
|
||||
// If we're at the beginning of the inputSource then we stop
|
||||
if (_currentPosition < 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
// Get more window if we need to
|
||||
if (!await _updateWindow()) {
|
||||
// Window is not available or things are empty so this
|
||||
// queue can't work right now
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// The current position should be equal to the first message to process
|
||||
// and the current window to process should not be empty
|
||||
return _inputSource.currentWindow.elements.isNotEmpty;
|
||||
}
|
||||
|
||||
// Slide the window toward the current position and load the batch around it
|
||||
Future<bool> updateWindow() async {
|
||||
|
||||
// Check if we are still in the window
|
||||
if (currentPosition>=windowFirst && currentPosition <= windowLast) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Get the length of the cubit
|
||||
final inputLength = await inputSource.cubit.operate((r) async => r.length);
|
||||
|
||||
// If not, slide the window
|
||||
if (currentPosition<windowFirst) {
|
||||
// Slide it backward, current position is now windowLast
|
||||
windowFirst = max((currentPosition - windowLength) + 1, 0);
|
||||
windowLast = currentPosition;
|
||||
} else {
|
||||
// Slide it forward, current position is now windowFirst
|
||||
windowFirst = currentPosition;
|
||||
windowLast = min((currentPosition + windowLength) - 1, inputLength - 1);
|
||||
}
|
||||
|
||||
// Get another input batch futher back
|
||||
final nextWindow =
|
||||
await inputSource.cubit.loadElements(windowLast + 1, (windowLast + 1) - windowFirst);
|
||||
final asErr = nextWindow.asError;
|
||||
if (asErr != null) {
|
||||
onError(asErr.error, asErr.stackTrace);
|
||||
return false;
|
||||
}
|
||||
final asLoading = nextWindow.asLoading;
|
||||
if (asLoading != null) {
|
||||
// xxx: no need to block the cubit here for this
|
||||
// xxx: might want to switch to a 'busy' state though
|
||||
// xxx: to let the messages view show a spinner at the bottom
|
||||
// xxx: while we reconcile...
|
||||
// emit(const AsyncValue.loading());
|
||||
return false;
|
||||
}
|
||||
currentWindow = nextWindow.asData!.value;
|
||||
Future<bool> _updateWindow() async {
|
||||
// Check if we are still in the window
|
||||
if (_currentPosition >= _inputSource.currentWindow.first &&
|
||||
_currentPosition <= _inputSource.currentWindow.last) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Get another input batch futher back
|
||||
final avOk =
|
||||
await _inputSource.updateWindow(_currentPosition, _maxWindowLength);
|
||||
|
||||
final asErr = avOk.asError;
|
||||
if (asErr != null) {
|
||||
_onError(asErr.error, asErr.stackTrace);
|
||||
return false;
|
||||
}
|
||||
final asLoading = avOk.asLoading;
|
||||
if (asLoading != null) {
|
||||
// xxx: no need to block the cubit here for this
|
||||
// xxx: might want to switch to a 'busy' state though
|
||||
// xxx: to let the messages view show a spinner at the bottom
|
||||
// xxx: while we reconcile...
|
||||
// emit(const AsyncValue.loading());
|
||||
return false;
|
||||
}
|
||||
return avOk.asData!.value;
|
||||
}
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
final TypedKey author;
|
||||
final ListQueue<proto.Message> toReconcile = ListQueue<proto.Message>();
|
||||
final AuthorInputSource inputSource;
|
||||
final OutputPosition? lastOutputPosition;
|
||||
final void Function(Object, StackTrace?) onError;
|
||||
final TypedKey _author;
|
||||
final AuthorInputSource _inputSource;
|
||||
final OutputPosition? _outputPosition;
|
||||
final void Function(Object, StackTrace?) _onError;
|
||||
final MessageIntegrity _messageIntegrity;
|
||||
|
||||
// The last message we've consumed
|
||||
proto.Message? _lastMessage;
|
||||
// The current position in the input log that we are looking at
|
||||
int currentPosition;
|
||||
// The current input window elements
|
||||
IList<DHTLogElementState<proto.Message>> currentWindow;
|
||||
// The first position of the sliding input window
|
||||
int windowFirst;
|
||||
// The last position of the sliding input window
|
||||
int windowLast;
|
||||
int _currentPosition;
|
||||
// The current message we're looking at
|
||||
proto.Message? _currentMessage;
|
||||
// Desired maximum window length
|
||||
int windowLength;
|
||||
|
||||
static const int _maxQueueChunk = 256;
|
||||
static const int _maxWindowLength = 256;
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue