Any way to configure writes to process watches synchronously?

[misha]

Hi!

Part of my application requires synchronous access to a part of the database - a snapshot - which I'd like to keep up-to-date automatically using watch. An update rolls in, I store the data, later code can use it immediately.

However, I noticed that after a write, stream processing is only performed during the next microtask delay. This defeats the purpose of automating snapshot generation with watch. Here's a code example of what I mean:

import 'package:drift/drift.dart';
import 'package:drift/native.dart';
import 'package:test/test.dart';

part 'drift_test.g.dart';

class Items extends Table {
  IntColumn get id => integer().autoIncrement()();
}

@DriftDatabase(tables: [Items])
class Database extends _$Database {
  Database() : super(NativeDatabase.memory());

  @override
  int get schemaVersion => 1;
}

void main() {
  test('watch updates depends on a microtask delay', () async {
    final db = Database();
    var triggered = false;
    db.items.all().watch().listen((_) => triggered = true);
    await db.into(db.items).insert(ItemsCompanion.insert());
    expect(triggered, isFalse);
    await Future.delayed(Duration.zero);
    expect(triggered, isTrue);
  });
}

As the test indicates, await does not imply that watchers have finished processing in any way.

My questions are as follows:

1. Is there a way to configure Drift to run watch processing synchronously with writes, thus allowing the first expect to be true in the test above?
2. Is there a different, canonical way to maintain a synchronous snapshot of a Drift database?

Unfortunately I haven't found anything after reading the (fantastic) documentation and many issues/discussions on GitHub.

Lastly, in case this is actually a platform-related issue, I am running on OpenSuse Tumbleweed.

Cheers,
Misha

[misha]

As frequently happens, I did discover a way to enforce this behavior using a QueryInterceptor shortly after posting this topic.

class MicrotaskDelayQueryInterceptor extends QueryInterceptor {
  @override
  Future<int> runInsert(QueryExecutor executor, String statement, List<Object?> args) async {
    final result = await super.runInsert(executor, statement, args);
    await Future.delayed(Duration.zero);
    return result;
  }

  // Same for updates, deletes, etc.
}

This accomplishes what I had in mind. Any drawbacks to this?

Edit: My success appears to be short-lived. While the delay does appear to allow insert streams to process, it does not work the same for runUpdate and runRemove. The reason is that insert includes notifyUpdates in its withCurrentExecutor block, whereas update and replace do not - they call notifyUpdates once withCurrentExecutor closes.

[simolus3]

I've added a bunch of print statements internally in drift to figure out where the delay is coming from, and it looks like this might be outside of our control.

First, I've adopted the test to first wait for the stream to emit its initial item before making the update:

import 'dart:async';

import 'package:drift/drift.dart';
import 'package:drift/native.dart';
import 'package:test/test.dart';

import '../generated/todos.dart';

void main() {
  test('watch updates depends on a microtask delay', () async {
    final db = TodoDb(NativeDatabase.memory());
    await db.customSelect('SELECT 1').get();

    var triggered = false;
    var firstCompleted = Completer<void>();
    db.todosTable.all().watch().listen((_) {
      if (!firstCompleted.isCompleted) {
        return firstCompleted.complete();
      }
      triggered = true;
    });

    await firstCompleted.future;

    await db
        .into(db.todosTable)
        .insert(TodosTableCompanion.insert(content: 'test content'));
    expect(triggered, isFalse);
    print('before yield');
    await Future.delayed(Duration.zero);
    print('after yield');
    expect(triggered, isTrue);
  });
}

Then with debug print in drift, I was able to verify that this line executes before we hit the await Future.delayed. In other words, all the data is already there, the stream subscription just has an inherent microtask delay built into it.

But also, it's worth noting that streams and updates are fundamentally different and independent asynchronous tasks. There is no guarantee that awaiting a write waits for all streams to have updated.

[misha]

Thank you looking into this.

In other words, all the data is already there, the stream subscription just has an inherent microtask delay built into it.

I suspected this was the case. I had a vision of being able make to asynchronous mutations, then turn around and read synchronously on the next line, but it turns out it's difficult to make those guarantees. I will go for explicit snapshots where necessary instead.