perlsky is a Perl 5 implementation of an AT Protocol Personal Data Server.
Base OutdatedCursor on event age
+36
-15
+7
-3
lib/ATProto/PDS/API/Util.pm
+7
-3
lib/ATProto/PDS/API/Util.pm
···
78
78
sub subscription_start_seq ($c, %args) {
79
79
my $cursor_param = $args{cursor_param};
80
80
my $latest = $args{latest} // $c->store->latest_event_seq;
81
-
my $oldest = $args{oldest} // $c->store->oldest_event_seq;
82
81
my $future_limit = $args{future_limit} // $latest;
83
82
my $future_message = $args{future_message} // 'Cursor is ahead of the local event stream';
84
83
my $outdated_message = $args{outdated_message} // 'Cursor predates the oldest locally retained event';
84
+
my $backfill_window = $args{backfill_window_seconds}
85
+
// $c->config_value('subscription_backfill_window_seconds', 3600);
86
+
my $backfill_cutoff = $args{backfill_cutoff} // (time - $backfill_window);
85
87
86
88
if (!defined $cursor_param || $cursor_param eq q()) {
87
89
return $latest + 1;
···
97
99
return undef;
98
100
}
99
101
100
-
if ($oldest && $cursor && $cursor < $oldest) {
102
+
my $next_event = $args{next_event} // $c->store->next_event_after_seq($cursor);
103
+
if ($next_event && ($next_event->{created_at} // 0) < $backfill_cutoff) {
101
104
$c->subscription_send(
102
105
binary => encode_info_frame('OutdatedCursor', $outdated_message),
103
106
frame_type => 'info',
104
107
);
105
-
return $oldest;
108
+
my $resume_seq = $args{backfill_start} // $c->store->earliest_event_seq_after_time($backfill_cutoff);
109
+
return defined($resume_seq) ? $resume_seq : ($latest + 1);
106
110
}
107
111
108
112
return $cursor + 1;
+19
-4
lib/ATProto/PDS/Store/SQLite.pm
+19
-4
lib/ATProto/PDS/Store/SQLite.pm
···
974
974
});
975
975
}
976
976
977
+
sub next_event_after_seq ($self, $cursor) {
978
+
return observe_store_operation($self->{metrics}, 'next_event_after_seq', sub {
979
+
return _row_from_blob_columns(_row_from_json_columns(
980
+
$self->dbh->selectrow_hashref(
981
+
q{SELECT * FROM events WHERE seq > ? ORDER BY seq LIMIT 1},
982
+
undef,
983
+
$cursor // 0,
984
+
),
985
+
qw(payload_json),
986
+
), qw(car_bytes));
987
+
});
988
+
}
989
+
977
990
sub latest_event_seq ($self) {
978
991
return observe_store_operation($self->{metrics}, 'latest_event_seq', sub {
979
992
return $self->dbh->selectrow_array(
···
982
995
});
983
996
}
984
997
985
-
sub oldest_event_seq ($self) {
986
-
return observe_store_operation($self->{metrics}, 'oldest_event_seq', sub {
998
+
sub earliest_event_seq_after_time ($self, $created_at) {
999
+
return observe_store_operation($self->{metrics}, 'earliest_event_seq_after_time', sub {
987
1000
my $value = $self->dbh->selectrow_array(
988
-
q{SELECT MIN(seq) FROM events},
1001
+
q{SELECT MIN(seq) FROM events WHERE created_at >= ?},
1002
+
undef,
1003
+
$created_at,
989
1004
);
990
-
return defined $value ? $value : 0;
1005
+
return $value;
991
1006
});
992
1007
}
993
1008
+5
-4
t/firehose.t
+5
-4
t/firehose.t
···
533
533
$skip_unknown->finish_ok;
534
534
535
535
my $outdated_floor = $app->store->latest_event_seq;
536
-
$app->store->dbh->do(q{DELETE FROM events WHERE seq <= ?}, undef, $outdated_floor);
536
+
my $stale_event_time = time - 3660;
537
+
$app->store->dbh->do(q{UPDATE events SET created_at = ? WHERE seq <= ?}, undef, $stale_event_time, $outdated_floor);
537
538
538
539
$t->post_ok('/xrpc/com.atproto.repo.createRecord' => {
539
540
Authorization => "Bearer $access",
···
557
558
is($outdated_info->{header}{t}, '#info', 'stale repo cursor yields an info frame');
558
559
is($outdated_info->{body}{name}, 'OutdatedCursor', 'stale repo cursor is reported as OutdatedCursor');
559
560
560
-
$outdated->message_ok('stale repo cursor then resumes from the oldest retained event');
561
+
$outdated->message_ok('stale repo cursor then resumes from the earliest event still inside the backfill window');
561
562
my $outdated_commit = decode_frame($outdated->message->[1]);
562
-
is($outdated_commit->{header}{t}, '#commit', 'repo stream resumes with the retained commit event');
563
-
is($outdated_commit->{body}{ops}[0]{path}, 'app.bsky.feed.post/firehose-fourth', 'repo replay resumes at the retained commit');
563
+
is($outdated_commit->{header}{t}, '#commit', 'repo stream resumes with the first event still inside the backfill window');
564
+
is($outdated_commit->{body}{ops}[0]{path}, 'app.bsky.feed.post/firehose-fourth', 'repo replay resumes at the first in-window commit');
564
565
$outdated->finish_ok;
565
566
566
567
done_testing;
+5
-4
t/labels.t
+5
-4
t/labels.t
···
359
359
ok($repo_neg_label, 'repo query includes the negated repo label itself');
360
360
is($repo_neg_label->{src}, $service_did, 'negated repo label keeps the local source');
361
361
362
-
$app->store->dbh->do(q{DELETE FROM events WHERE seq <= ?}, undef, $app->store->latest_event_seq);
362
+
my $stale_label_time = time - 3660;
363
+
$app->store->dbh->do(q{UPDATE events SET created_at = ? WHERE seq <= ?}, undef, $stale_label_time, $app->store->latest_event_seq);
363
364
364
365
$t->post_ok('/xrpc/com.atproto.admin.updateSubjectStatus' => {
365
366
Authorization => $admin_auth,
···
376
377
is($outdated_info->{header}{t}, '#info', 'stale label cursor yields an info frame');
377
378
is($outdated_info->{body}{name}, 'OutdatedCursor', 'stale label cursor is reported as OutdatedCursor');
378
379
379
-
$outdated->message_ok('stale label cursor then resumes from the oldest retained label event');
380
+
$outdated->message_ok('stale label cursor then resumes from the earliest in-window label event');
380
381
my $outdated_label = decode_frame($outdated->message->[1]);
381
382
is($outdated_label->{header}{t}, '#labels', 'label stream resumes with a labels frame');
382
-
is($outdated_label->{body}{labels}[0]{uri}, "at://$did", 'stale label replay resumes at the retained label event');
383
-
is($outdated_label->{body}{labels}[0]{val}, '!hide', 'retained label replay carries the expected moderation label');
383
+
is($outdated_label->{body}{labels}[0]{uri}, "at://$did", 'stale label replay resumes at the first in-window label event');
384
+
is($outdated_label->{body}{labels}[0]{val}, '!hide', 'first in-window label replay carries the expected moderation label');
384
385
$outdated->finish_ok;
385
386
386
387
$ws->finish_ok;