perlsky is a Perl 5 implementation of an AT Protocol Personal Data Server.
Add reference PDS differential validation
+908
+618
script/differential-validate
+618
script/differential-validate
···
1
+
#!/usr/bin/env perl
2
+
use v5.34;
3
+
use warnings;
4
+
use feature 'signatures';
5
+
no warnings 'experimental::signatures';
6
+
7
+
use Config ();
8
+
use File::Basename qw(dirname);
9
+
use File::Path qw(make_path);
10
+
use File::Spec;
11
+
use File::Temp qw(tempdir);
12
+
use IO::Socket::INET;
13
+
use JSON::PP ();
14
+
use POSIX qw(WNOHANG);
15
+
use Time::HiRes qw(sleep time);
16
+
17
+
BEGIN {
18
+
require lib;
19
+
my $root = File::Spec->rel2abs(File::Spec->catdir(dirname(__FILE__), '..'));
20
+
lib->import(
21
+
File::Spec->catdir($root, 'lib'),
22
+
File::Spec->catdir($root, 'local', 'lib', 'perl5'),
23
+
File::Spec->catdir($root, 'local', 'lib', 'perl5', $Config::Config{archname}),
24
+
);
25
+
}
26
+
27
+
use Mojo::JSON qw(decode_json encode_json true false);
28
+
use Mojo::URL;
29
+
use Mojo::UserAgent;
30
+
use ATProto::PDS::EventStream qw(decode_frame);
31
+
32
+
my $root = File::Spec->rel2abs(File::Spec->catdir(dirname(__FILE__), '..'));
33
+
my $tmp = tempdir(CLEANUP => 1);
34
+
my @children;
35
+
my $failed = 0;
36
+
37
+
END {
38
+
my $status = $?;
39
+
for my $child (reverse @children) {
40
+
next unless $child->{pid};
41
+
my $alive = kill 0, $child->{pid};
42
+
next unless $alive;
43
+
kill 'TERM', $child->{pid};
44
+
for (1 .. 40) {
45
+
my $done = waitpid($child->{pid}, WNOHANG);
46
+
last if $done == $child->{pid};
47
+
sleep 0.1;
48
+
}
49
+
kill 'KILL', $child->{pid} if kill 0, $child->{pid};
50
+
waitpid($child->{pid}, 0);
51
+
}
52
+
$? = $status;
53
+
}
54
+
55
+
sub note ($message) {
56
+
print "$message\n";
57
+
}
58
+
59
+
sub pass ($message) {
60
+
print "ok - $message\n";
61
+
}
62
+
63
+
sub fail_check ($message) {
64
+
print "not ok - $message\n";
65
+
$failed++;
66
+
}
67
+
68
+
sub check ($condition, $message) {
69
+
if ($condition) {
70
+
pass($message);
71
+
} else {
72
+
fail_check($message);
73
+
}
74
+
}
75
+
76
+
sub free_port () {
77
+
my $sock = IO::Socket::INET->new(
78
+
LocalAddr => '127.0.0.1',
79
+
LocalPort => 0,
80
+
Proto => 'tcp',
81
+
Listen => 5,
82
+
ReuseAddr => 1,
83
+
) or die "unable to allocate a free port: $!";
84
+
my $port = $sock->sockport;
85
+
close $sock;
86
+
return $port;
87
+
}
88
+
89
+
sub spawn_logged ($name, $cmd, $env, $log_path) {
90
+
open my $log_fh, '>', $log_path or die "unable to open $log_path: $!";
91
+
my $pid = fork;
92
+
die "unable to fork for $name: $!" unless defined $pid;
93
+
94
+
if ($pid == 0) {
95
+
chdir $root or die "unable to chdir to $root: $!";
96
+
open STDOUT, '>&', $log_fh or die "unable to redirect stdout for $name: $!";
97
+
open STDERR, '>&', $log_fh or die "unable to redirect stderr for $name: $!";
98
+
%ENV = (%ENV, %{$env // {}});
99
+
exec @{$cmd} or die "unable to exec $name: $!";
100
+
}
101
+
102
+
close $log_fh;
103
+
my $child = {
104
+
name => $name,
105
+
pid => $pid,
106
+
log => $log_path,
107
+
};
108
+
push @children, $child;
109
+
return $child;
110
+
}
111
+
112
+
sub slurp_file ($path) {
113
+
return q() unless -f $path;
114
+
open my $fh, '<', $path or return q();
115
+
local $/;
116
+
return <$fh> // q();
117
+
}
118
+
119
+
sub wait_for_ready_file ($name, $path, $timeout = 30) {
120
+
my $deadline = time + $timeout;
121
+
while (time < $deadline) {
122
+
if (-f $path) {
123
+
return decode_json(slurp_file($path));
124
+
}
125
+
sleep 0.1;
126
+
}
127
+
die "timed out waiting for $name ready file $path\n";
128
+
}
129
+
130
+
sub wait_for_http_ok ($name, $url, $timeout = 30) {
131
+
my $ua = Mojo::UserAgent->new(max_redirects => 0);
132
+
my $deadline = time + $timeout;
133
+
while (time < $deadline) {
134
+
my $tx = eval { $ua->get($url) };
135
+
if ($tx) {
136
+
my $res = eval { $tx->result };
137
+
return $res if $res && $res->is_success;
138
+
}
139
+
sleep 0.2;
140
+
}
141
+
die "timed out waiting for $name at $url\n";
142
+
}
143
+
144
+
sub setup_reference_runtime () {
145
+
my @cmd = ($^X, File::Spec->catfile($root, 'script', 'setup-reference-runtime'));
146
+
system(@cmd) == 0 or die "reference runtime setup failed\n";
147
+
}
148
+
149
+
sub random_hex ($bytes) {
150
+
open my $fh, '<:raw', '/dev/urandom' or die "unable to read /dev/urandom: $!";
151
+
read($fh, my $buf, $bytes) == $bytes or die "unable to read $bytes bytes from /dev/urandom\n";
152
+
close $fh;
153
+
return unpack('H*', $buf);
154
+
}
155
+
156
+
sub post_json ($origin, $nsid, $payload, $headers = {}) {
157
+
my $ua = Mojo::UserAgent->new(max_redirects => 0);
158
+
my $tx = $ua->post(
159
+
"$origin/xrpc/$nsid" => {
160
+
'Content-Type' => 'application/json',
161
+
%{$headers},
162
+
} => json => $payload,
163
+
);
164
+
return $tx->result;
165
+
}
166
+
167
+
sub get_form ($origin, $nsid, $query, $headers = {}) {
168
+
my $ua = Mojo::UserAgent->new(max_redirects => 0);
169
+
my $url = Mojo::URL->new("$origin/xrpc/$nsid")->query($query);
170
+
my $tx = $ua->get($url => $headers);
171
+
return $tx->result;
172
+
}
173
+
174
+
sub get_json ($origin, $nsid, $query = undef, $headers = {}) {
175
+
my $res = defined $query ? get_form($origin, $nsid, $query, $headers) : Mojo::UserAgent->new(max_redirects => 0)->get("$origin/xrpc/$nsid" => $headers)->result;
176
+
return $res;
177
+
}
178
+
179
+
sub auth_header ($token) {
180
+
return { Authorization => "Bearer $token" };
181
+
}
182
+
183
+
sub normalized_domains ($res) {
184
+
my $json = $res->json || {};
185
+
return [ sort map { /^\./ ? $_ : ".$_" } @{ $json->{availableUserDomains} || [] } ];
186
+
}
187
+
188
+
sub quiet_firehose ($url, $quiet_seconds = 0.5) {
189
+
my $ua = Mojo::UserAgent->new;
190
+
my $got_frame = 0;
191
+
my $error;
192
+
my $done = 0;
193
+
194
+
$ua->websocket($url => sub ($ua, $tx) {
195
+
unless ($tx->is_websocket) {
196
+
$error = $tx->res->error->{message} // 'websocket handshake failed';
197
+
$done = 1;
198
+
Mojo::IOLoop->stop;
199
+
return;
200
+
}
201
+
202
+
my $timer = Mojo::IOLoop->timer($quiet_seconds => sub {
203
+
$done = 1;
204
+
$tx->finish(1000);
205
+
});
206
+
207
+
$tx->on(binary => sub ($tx, $bytes) {
208
+
$got_frame = 1;
209
+
Mojo::IOLoop->remove($timer);
210
+
$tx->finish(1000);
211
+
});
212
+
213
+
$tx->on(finish => sub ($tx, $code, $reason = undef) {
214
+
Mojo::IOLoop->stop if $done || $got_frame || defined $error;
215
+
});
216
+
});
217
+
218
+
Mojo::IOLoop->start unless Mojo::IOLoop->is_running;
219
+
die "$error\n" if defined $error;
220
+
return !$got_frame;
221
+
}
222
+
223
+
sub next_commit_frame ($url, $expected_path, $trigger, $timeout = 10) {
224
+
my $ua = Mojo::UserAgent->new;
225
+
my $frame;
226
+
my $error;
227
+
my $deadline = time + $timeout;
228
+
my $triggered = 0;
229
+
my $done = 0;
230
+
231
+
$ua->websocket($url => sub ($ua, $tx) {
232
+
unless ($tx->is_websocket) {
233
+
$error = $tx->res->error->{message} // 'websocket handshake failed';
234
+
$done = 1;
235
+
Mojo::IOLoop->stop;
236
+
return;
237
+
}
238
+
239
+
my $timer;
240
+
$timer = Mojo::IOLoop->recurring(0.1 => sub {
241
+
if (time >= $deadline) {
242
+
$error = "timed out waiting for firehose commit at $expected_path";
243
+
Mojo::IOLoop->remove($timer);
244
+
$done = 1;
245
+
$tx->finish(1000);
246
+
}
247
+
});
248
+
249
+
$tx->on(binary => sub ($tx, $bytes) {
250
+
my $decoded = decode_frame($bytes);
251
+
my $header = $decoded->{header} || {};
252
+
my $body = $decoded->{body} || {};
253
+
my @ops = @{ $body->{ops} || [] };
254
+
my $match = grep { ($_->{path} // q()) eq $expected_path } @ops;
255
+
if (($header->{t} // q()) eq '#commit' && $match) {
256
+
$frame = $decoded;
257
+
Mojo::IOLoop->remove($timer);
258
+
$done = 1;
259
+
$tx->finish(1000);
260
+
}
261
+
});
262
+
263
+
Mojo::IOLoop->next_tick(sub {
264
+
return if $triggered;
265
+
$triggered = 1;
266
+
eval { $trigger->() };
267
+
if ($@) {
268
+
$error = $@;
269
+
Mojo::IOLoop->remove($timer);
270
+
$done = 1;
271
+
$tx->finish(1011);
272
+
}
273
+
});
274
+
275
+
$tx->on(finish => sub ($tx, $code, $reason = undef) {
276
+
Mojo::IOLoop->stop if $done || defined $error;
277
+
});
278
+
});
279
+
280
+
Mojo::IOLoop->start unless Mojo::IOLoop->is_running;
281
+
die "$error\n" if defined $error;
282
+
return $frame;
283
+
}
284
+
285
+
sub first_frame ($url, $timeout = 5) {
286
+
my $ua = Mojo::UserAgent->new;
287
+
my $frame;
288
+
my $error;
289
+
my $done = 0;
290
+
my $deadline = time + $timeout;
291
+
292
+
$ua->websocket($url => sub ($ua, $tx) {
293
+
unless ($tx->is_websocket) {
294
+
$error = $tx->res->error->{message} // 'websocket handshake failed';
295
+
$done = 1;
296
+
Mojo::IOLoop->stop;
297
+
return;
298
+
}
299
+
300
+
my $timer;
301
+
$timer = Mojo::IOLoop->recurring(0.1 => sub {
302
+
if (time >= $deadline) {
303
+
$error = "timed out waiting for firehose frame";
304
+
Mojo::IOLoop->remove($timer);
305
+
$done = 1;
306
+
$tx->finish(1000);
307
+
}
308
+
});
309
+
310
+
$tx->on(binary => sub ($tx, $bytes) {
311
+
$frame = decode_frame($bytes);
312
+
Mojo::IOLoop->remove($timer);
313
+
$done = 1;
314
+
$tx->finish(1000);
315
+
});
316
+
317
+
$tx->on(finish => sub ($tx, $code, $reason = undef) {
318
+
Mojo::IOLoop->stop if $done || defined $error;
319
+
});
320
+
});
321
+
322
+
Mojo::IOLoop->start unless Mojo::IOLoop->is_running;
323
+
die "$error\n" if defined $error;
324
+
return $frame;
325
+
}
326
+
327
+
sub normalize_commit_frame ($frame, $did) {
328
+
my $header = $frame->{header} || {};
329
+
my $body = $frame->{body} || {};
330
+
my $first = $body->{ops}[0] || {};
331
+
332
+
return {
333
+
op => $header->{op},
334
+
type => $header->{t},
335
+
repo_match => (($body->{repo} // q()) eq $did) ? 1 : 0,
336
+
action => $first->{action},
337
+
path => $first->{path},
338
+
has_blocks => length($body->{blocks} // q()) > 0 ? 1 : 0,
339
+
has_commit => defined $body->{commit} ? 1 : 0,
340
+
seq_is_int => defined($body->{seq}) && $body->{seq} =~ /\A\d+\z/ ? 1 : 0,
341
+
};
342
+
}
343
+
344
+
sub same_hash ($left, $right) {
345
+
my $json = JSON::PP->new->canonical(1)->allow_nonref(1);
346
+
return $json->encode($left) eq $json->encode($right);
347
+
}
348
+
349
+
note('Preparing official reference runtime');
350
+
setup_reference_runtime();
351
+
352
+
my $plc_port = free_port();
353
+
my $reference_port = free_port();
354
+
my $perl_port = free_port();
355
+
356
+
my $plc_ready = File::Spec->catfile($tmp, 'plc.ready.json');
357
+
my $ref_ready = File::Spec->catfile($tmp, 'reference.ready.json');
358
+
359
+
my $plc_log = File::Spec->catfile($tmp, 'plc.log');
360
+
my $ref_log = File::Spec->catfile($tmp, 'reference.log');
361
+
my $perl_log = File::Spec->catfile($tmp, 'perlds.log');
362
+
363
+
my $plc = spawn_logged(
364
+
'plc-mock',
365
+
['fnm', 'exec', '--using=20', '--', 'node', File::Spec->catfile($root, 'tools', 'differential', 'plc-mock.cjs')],
366
+
{
367
+
PERLDS_READY_FILE => $plc_ready,
368
+
PERLDS_PLC_PORT => $plc_port,
369
+
PERLDS_PLC_HOST => '127.0.0.1',
370
+
},
371
+
$plc_log,
372
+
);
373
+
374
+
my $plc_info = wait_for_ready_file('plc mock', $plc_ready);
375
+
pass("started local PLC mock at $plc_info->{origin}");
376
+
377
+
my $reference_data = File::Spec->catdir($tmp, 'reference');
378
+
make_path($reference_data);
379
+
380
+
my $reference = spawn_logged(
381
+
'reference-pds',
382
+
['fnm', 'exec', '--using=20', '--', 'node', File::Spec->catfile($root, 'tools', 'differential', 'reference-pds-runner.cjs')],
383
+
{
384
+
PERLDS_READY_FILE => $ref_ready,
385
+
PDS_PORT => $reference_port,
386
+
PDS_HOSTNAME => 'localhost',
387
+
PDS_DEV_MODE => 1,
388
+
PDS_DATA_DIRECTORY => File::Spec->catdir($reference_data, 'data'),
389
+
PDS_BLOBSTORE_DISK_LOCATION => File::Spec->catdir($reference_data, 'blobs'),
390
+
PDS_BLOBSTORE_DISK_TMP_LOCATION => File::Spec->catdir($reference_data, 'blobs-tmp'),
391
+
PDS_PLC_ROTATION_KEY_K256_PRIVATE_KEY_HEX => random_hex(32),
392
+
PDS_JWT_SECRET => 'reference-jwt-secret',
393
+
PDS_ADMIN_PASSWORD => 'reference-admin-secret',
394
+
PDS_INVITE_REQUIRED => 0,
395
+
PDS_DID_PLC_URL => $plc_info->{origin},
396
+
PDS_CONTACT_EMAIL_ADDRESS => 'abuse@example.test',
397
+
},
398
+
$ref_log,
399
+
);
400
+
401
+
my $reference_info = wait_for_ready_file('reference pds', $ref_ready);
402
+
wait_for_http_ok('reference pds', "$reference_info->{origin}/xrpc/_health");
403
+
pass("started official reference PDS at $reference_info->{origin}");
404
+
405
+
my $perlds_config = File::Spec->catfile($tmp, 'perlds.json');
406
+
open my $cfg_fh, '>', $perlds_config or die "unable to write $perlds_config: $!";
407
+
print {$cfg_fh} encode_json({
408
+
base_url => "http://127.0.0.1:$perl_port",
409
+
service_handle_domain => 'test',
410
+
service_did_method => 'did:web',
411
+
jwt_secret => 'perlds-jwt-secret',
412
+
admin_password => 'perlds-admin-secret',
413
+
data_dir => File::Spec->catdir($tmp, 'perlds-data'),
414
+
db_path => File::Spec->catfile($tmp, 'perlds.sqlite'),
415
+
});
416
+
close $cfg_fh;
417
+
418
+
my $perlds = spawn_logged(
419
+
'perlds',
420
+
[$^X, File::Spec->catfile($root, 'script', 'perlds'), 'daemon', '-l', "http://127.0.0.1:$perl_port"],
421
+
{
422
+
PERLDS_CONFIG => $perlds_config,
423
+
},
424
+
$perl_log,
425
+
);
426
+
427
+
wait_for_http_ok('perlds', "http://127.0.0.1:$perl_port/_health");
428
+
pass("started perlds at http://127.0.0.1:$perl_port");
429
+
430
+
my %server = (
431
+
reference => {
432
+
origin => $reference_info->{origin},
433
+
handle => 'alice.test',
434
+
email => 'alice-ref@test.com',
435
+
},
436
+
perlds => {
437
+
origin => "http://127.0.0.1:$perl_port",
438
+
handle => 'alice.test',
439
+
email => 'alice-perl@test.com',
440
+
},
441
+
);
442
+
443
+
note('Comparing describeServer');
444
+
for my $name (sort keys %server) {
445
+
my $res = get_json($server{$name}{origin}, 'com.atproto.server.describeServer');
446
+
check($res->is_success, "$name describeServer succeeds");
447
+
$server{$name}{describe} = $res->json if $res->is_success;
448
+
}
449
+
450
+
check(
451
+
same_hash(
452
+
{
453
+
inviteCodeRequired => $server{reference}{describe}{inviteCodeRequired} ? true : false,
454
+
availableUserDomains => normalized_domains(get_json($server{reference}{origin}, 'com.atproto.server.describeServer')),
455
+
},
456
+
{
457
+
inviteCodeRequired => $server{perlds}{describe}{inviteCodeRequired} ? true : false,
458
+
availableUserDomains => normalized_domains(get_json($server{perlds}{origin}, 'com.atproto.server.describeServer')),
459
+
},
460
+
),
461
+
'describeServer matches on invite requirement and user domains',
462
+
);
463
+
464
+
note('Creating matching accounts');
465
+
for my $name (sort keys %server) {
466
+
my $res = post_json($server{$name}{origin}, 'com.atproto.server.createAccount', {
467
+
handle => $server{$name}{handle},
468
+
email => $server{$name}{email},
469
+
password => 'hunter22',
470
+
});
471
+
check($res->is_success, "$name createAccount succeeds");
472
+
next unless $res->is_success;
473
+
474
+
my $json = $res->json;
475
+
$server{$name}{did} = $json->{did};
476
+
$server{$name}{access} = $json->{accessJwt};
477
+
$server{$name}{refresh} = $json->{refreshJwt};
478
+
479
+
check(($json->{handle} // q()) eq $server{$name}{handle}, "$name createAccount returns normalized handle");
480
+
check(defined $json->{did} && $json->{did} =~ /\Adid:/, "$name createAccount returns a DID");
481
+
check(defined $json->{accessJwt} && length $json->{accessJwt}, "$name createAccount returns an access token");
482
+
check(defined $json->{refreshJwt} && length $json->{refreshJwt}, "$name createAccount returns a refresh token");
483
+
}
484
+
485
+
note('Comparing resolveHandle');
486
+
for my $name (sort keys %server) {
487
+
my $res = get_form($server{$name}{origin}, 'com.atproto.identity.resolveHandle', { handle => $server{$name}{handle} });
488
+
check($res->is_success, "$name resolveHandle succeeds");
489
+
next unless $res->is_success;
490
+
check(($res->json->{did} // q()) eq $server{$name}{did}, "$name resolveHandle returns the created DID");
491
+
}
492
+
493
+
note('Comparing getSession');
494
+
for my $name (sort keys %server) {
495
+
my $res = get_json($server{$name}{origin}, 'com.atproto.server.getSession', undef, auth_header($server{$name}{access}));
496
+
check($res->is_success, "$name getSession succeeds");
497
+
next unless $res->is_success;
498
+
my $json = $res->json;
499
+
check(($json->{did} // q()) eq $server{$name}{did}, "$name getSession returns the created DID");
500
+
check(($json->{handle} // q()) eq $server{$name}{handle}, "$name getSession returns the created handle");
501
+
}
502
+
503
+
my $record = {
504
+
'$type' => 'app.bsky.feed.post',
505
+
text => 'differential validation post',
506
+
createdAt => '2026-03-10T00:00:00Z',
507
+
};
508
+
509
+
note('Comparing createRecord');
510
+
for my $name (sort keys %server) {
511
+
my $res = post_json($server{$name}{origin}, 'com.atproto.repo.createRecord', {
512
+
repo => $server{$name}{did},
513
+
collection => 'app.bsky.feed.post',
514
+
rkey => 'diffpost',
515
+
record => $record,
516
+
}, auth_header($server{$name}{access}));
517
+
check($res->is_success, "$name createRecord succeeds");
518
+
next unless $res->is_success;
519
+
my $json = $res->json;
520
+
$server{$name}{record_uri} = $json->{uri};
521
+
check(($json->{uri} // q()) =~ m{/app\.bsky\.feed\.post/diffpost\z}, "$name createRecord returns the expected record URI");
522
+
check(defined $json->{cid} && length $json->{cid}, "$name createRecord returns a CID");
523
+
}
524
+
525
+
note('Comparing listRecords');
526
+
for my $name (sort keys %server) {
527
+
my $res = get_form($server{$name}{origin}, 'com.atproto.repo.listRecords', {
528
+
repo => $server{$name}{did},
529
+
collection => 'app.bsky.feed.post',
530
+
});
531
+
check($res->is_success, "$name listRecords succeeds");
532
+
next unless $res->is_success;
533
+
my $records = $res->json->{records} || [];
534
+
check(@{$records} >= 1, "$name listRecords returns at least one record");
535
+
check(($records->[0]{value}{text} // q()) eq $record->{text}, "$name listRecords returns the created text");
536
+
}
537
+
538
+
note('Comparing getLatestCommit');
539
+
for my $name (sort keys %server) {
540
+
my $res = get_form($server{$name}{origin}, 'com.atproto.sync.getLatestCommit', { did => $server{$name}{did} });
541
+
check($res->is_success, "$name getLatestCommit succeeds");
542
+
next unless $res->is_success;
543
+
my $json = $res->json;
544
+
check(defined $json->{cid} && length $json->{cid}, "$name getLatestCommit returns a CID");
545
+
check(defined $json->{rev} && length $json->{rev}, "$name getLatestCommit returns a rev");
546
+
}
547
+
548
+
note('Comparing getRepo CAR exports');
549
+
for my $name (sort keys %server) {
550
+
my $res = get_form($server{$name}{origin}, 'com.atproto.sync.getRepo', { did => $server{$name}{did} });
551
+
check($res->is_success, "$name getRepo succeeds");
552
+
next unless $res->is_success;
553
+
my $ctype = $res->headers->content_type // q();
554
+
check($ctype =~ m{application/vnd\.ipld\.car}, "$name getRepo returns CAR bytes");
555
+
check(length($res->body // q()) > 0, "$name getRepo CAR payload is non-empty");
556
+
}
557
+
558
+
note('Comparing firehose live follow behavior');
559
+
for my $name (sort keys %server) {
560
+
my $quiet = quiet_firehose("$server{$name}{origin}/xrpc/com.atproto.sync.subscribeRepos");
561
+
check($quiet, "$name subscribeRepos stays quiet with no cursor before a new write");
562
+
}
563
+
564
+
for my $name (sort keys %server) {
565
+
my $path = 'app.bsky.feed.post/firehose-diff';
566
+
my $frame = next_commit_frame(
567
+
"$server{$name}{origin}/xrpc/com.atproto.sync.subscribeRepos",
568
+
$path,
569
+
sub {
570
+
my $res = post_json($server{$name}{origin}, 'com.atproto.repo.createRecord', {
571
+
repo => $server{$name}{did},
572
+
collection => 'app.bsky.feed.post',
573
+
rkey => 'firehose-diff',
574
+
record => {
575
+
%{$record},
576
+
text => "firehose validation for $name",
577
+
},
578
+
}, auth_header($server{$name}{access}));
579
+
die "createRecord for firehose failed on $name\n" unless $res->is_success;
580
+
},
581
+
);
582
+
$server{$name}{firehose_commit} = normalize_commit_frame($frame, $server{$name}{did});
583
+
check($server{$name}{firehose_commit}{repo_match}, "$name firehose commit belongs to the created repo");
584
+
}
585
+
586
+
check(
587
+
same_hash($server{reference}{firehose_commit}, $server{perlds}{firehose_commit}),
588
+
'subscribeRepos emits the same normalized commit semantics as the official reference PDS',
589
+
);
590
+
591
+
note('Comparing future cursor error behavior');
592
+
for my $name (sort keys %server) {
593
+
my $frame = first_frame("$server{$name}{origin}/xrpc/com.atproto.sync.subscribeRepos?cursor=999999999999");
594
+
my $header = $frame->{header} || {};
595
+
my $body = $frame->{body} || {};
596
+
$server{$name}{future_cursor} = {
597
+
op => $header->{op},
598
+
error => $body->{error},
599
+
};
600
+
}
601
+
602
+
check(
603
+
same_hash($server{reference}{future_cursor}, $server{perlds}{future_cursor}),
604
+
'subscribeRepos agrees on future cursor errors',
605
+
);
606
+
607
+
if ($failed) {
608
+
print "\nReference PDS log:\n";
609
+
print slurp_file($ref_log);
610
+
print "\nperlds log:\n";
611
+
print slurp_file($perl_log);
612
+
print "\nPLC mock log:\n";
613
+
print slurp_file($plc_log);
614
+
die "\ndifferential validation failed with $failed mismatches\n";
615
+
}
616
+
617
+
note('Differential validation succeeded');
618
+
exit 0;
+38
script/setup-reference-runtime
+38
script/setup-reference-runtime
···
1
+
#!/usr/bin/env perl
2
+
use v5.34;
3
+
use warnings;
4
+
5
+
use File::Basename qw(dirname);
6
+
use File::Path qw(make_path);
7
+
use File::Spec;
8
+
9
+
my $root = File::Spec->rel2abs(File::Spec->catdir(dirname(__FILE__), '..'));
10
+
my $runtime_dir = File::Spec->catdir($root, '.tools', 'reference-runtime');
11
+
my $module_path = File::Spec->catdir($runtime_dir, 'node_modules', '@atproto', 'pds');
12
+
13
+
if (-d $module_path) {
14
+
print "$module_path\n";
15
+
exit 0;
16
+
}
17
+
18
+
make_path($runtime_dir);
19
+
20
+
chdir $runtime_dir or die "unable to chdir to $runtime_dir: $!";
21
+
22
+
if (!-f File::Spec->catfile($runtime_dir, 'package.json')) {
23
+
system('npm', 'init', '-y') == 0
24
+
or die "npm init failed\n";
25
+
}
26
+
27
+
my @fnm_check = ('fnm', 'exec', '--using=20', '--', 'node', '--version');
28
+
if (system(@fnm_check) != 0) {
29
+
system('fnm', 'install', '20') == 0
30
+
or die "unable to install Node 20 with fnm\n";
31
+
}
32
+
33
+
system(
34
+
'fnm', 'exec', '--using=20', '--',
35
+
'npm', 'install', '--no-save', '@atproto/pds@0.4.214',
36
+
) == 0 or die "npm install for \@atproto/pds failed\n";
37
+
38
+
print "$module_path\n";
+31
t/reference-differential.t
+31
t/reference-differential.t
···
1
+
use v5.34;
2
+
use warnings;
3
+
4
+
use Config ();
5
+
use FindBin qw($Bin);
6
+
use File::Spec;
7
+
use Test2::V0;
8
+
9
+
BEGIN {
10
+
require lib;
11
+
my $root = File::Spec->rel2abs(File::Spec->catdir($Bin, '..'));
12
+
lib->import(
13
+
File::Spec->catdir($root, 'lib'),
14
+
File::Spec->catdir($root, 'local', 'lib', 'perl5'),
15
+
File::Spec->catdir($root, 'local', 'lib', 'perl5', $Config::Config{archname}),
16
+
);
17
+
}
18
+
19
+
plan skip_all => 'set PERLDS_RUN_REFERENCE_DIFF=1 to run the official reference PDS differential harness'
20
+
unless $ENV{PERLDS_RUN_REFERENCE_DIFF};
21
+
22
+
my $root = File::Spec->rel2abs(File::Spec->catdir($Bin, '..'));
23
+
my $script = File::Spec->catfile($root, 'script', 'differential-validate');
24
+
25
+
my $output = qx{$^X $script 2>&1};
26
+
my $code = $? >> 8;
27
+
28
+
diag($output) if $code;
29
+
is($code, 0, 'reference differential harness exits successfully');
30
+
31
+
done_testing;
+154
tools/differential/plc-mock.cjs
+154
tools/differential/plc-mock.cjs
···
1
+
#!/usr/bin/env node
2
+
3
+
const fs = require('node:fs');
4
+
const http = require('node:http');
5
+
const path = require('node:path');
6
+
const { URL } = require('node:url');
7
+
const plc = require(path.join(
8
+
__dirname,
9
+
'..',
10
+
'..',
11
+
'.tools',
12
+
'reference-runtime',
13
+
'node_modules',
14
+
'@did-plc',
15
+
'lib',
16
+
'dist',
17
+
));
18
+
19
+
const readyFile = process.env.PERLDS_READY_FILE;
20
+
const port = Number(process.env.PERLDS_PLC_PORT || '0');
21
+
const host = process.env.PERLDS_PLC_HOST || '127.0.0.1';
22
+
23
+
if (!readyFile) {
24
+
console.error('PERLDS_READY_FILE is required');
25
+
process.exit(1);
26
+
}
27
+
28
+
const store = new Map();
29
+
30
+
const sendJson = (res, status, body) => {
31
+
const payload = JSON.stringify(body);
32
+
res.writeHead(status, {
33
+
'content-type': 'application/json',
34
+
'content-length': Buffer.byteLength(payload),
35
+
});
36
+
res.end(payload);
37
+
};
38
+
39
+
const notFound = (res) => {
40
+
sendJson(res, 404, { error: 'DidNotFound' });
41
+
};
42
+
43
+
const normalizePath = (pathname) => pathname.replace(/\/+$/, '') || '/';
44
+
45
+
const readBody = async (req) => {
46
+
const chunks = [];
47
+
for await (const chunk of req) {
48
+
chunks.push(chunk);
49
+
}
50
+
const raw = Buffer.concat(chunks).toString('utf8');
51
+
return raw ? JSON.parse(raw) : {};
52
+
};
53
+
54
+
const currentDocument = (did) => {
55
+
const entry = store.get(did);
56
+
if (!entry) {
57
+
return null;
58
+
}
59
+
const op = entry.ops[entry.ops.length - 1];
60
+
return plc.formatDidDoc({ did, ...op });
61
+
};
62
+
63
+
const currentData = (did) => {
64
+
const entry = store.get(did);
65
+
if (!entry) {
66
+
return null;
67
+
}
68
+
const op = entry.ops[entry.ops.length - 1];
69
+
return {
70
+
did,
71
+
rotationKeys: op.rotationKeys,
72
+
verificationMethods: op.verificationMethods,
73
+
alsoKnownAs: op.alsoKnownAs,
74
+
services: op.services,
75
+
};
76
+
};
77
+
78
+
const server = http.createServer(async (req, res) => {
79
+
try {
80
+
const url = new URL(req.url, `http://${req.headers.host || 'localhost'}`);
81
+
const path = normalizePath(url.pathname);
82
+
const parts = path.split('/').filter(Boolean).map(decodeURIComponent);
83
+
84
+
if (parts.length === 0) {
85
+
return sendJson(res, 200, { service: 'perlds-plc-mock' });
86
+
}
87
+
88
+
const did = parts[0];
89
+
90
+
if (req.method === 'GET' && parts.length === 1) {
91
+
const doc = currentDocument(did);
92
+
return doc ? sendJson(res, 200, doc) : notFound(res);
93
+
}
94
+
95
+
if (req.method === 'GET' && parts[1] === 'data' && parts.length === 2) {
96
+
const data = currentData(did);
97
+
return data ? sendJson(res, 200, data) : notFound(res);
98
+
}
99
+
100
+
if (req.method === 'GET' && parts[1] === 'log' && parts[2] === 'last' && parts.length === 3) {
101
+
const entry = store.get(did);
102
+
return entry ? sendJson(res, 200, entry.ops[entry.ops.length - 1]) : notFound(res);
103
+
}
104
+
105
+
if (req.method === 'GET' && parts[1] === 'log' && parts.length === 2) {
106
+
const entry = store.get(did);
107
+
return entry ? sendJson(res, 200, entry.ops) : notFound(res);
108
+
}
109
+
110
+
if (req.method === 'POST' && parts.length === 1) {
111
+
const op = await readBody(req);
112
+
await plc.assureValidOp(op);
113
+
114
+
const existing = store.get(did);
115
+
if (!existing && op.prev !== null) {
116
+
return sendJson(res, 400, { error: 'MissingPreviousOp' });
117
+
}
118
+
if (existing && op.prev === null) {
119
+
return sendJson(res, 400, { error: 'AlreadyExists' });
120
+
}
121
+
122
+
if (!existing) {
123
+
const computedDid = await plc.didForCreateOp(op);
124
+
if (computedDid !== did) {
125
+
return sendJson(res, 400, { error: 'DidMismatch' });
126
+
}
127
+
store.set(did, { ops: [op] });
128
+
return sendJson(res, 200, { ok: true });
129
+
}
130
+
131
+
existing.ops.push(op);
132
+
return sendJson(res, 200, { ok: true });
133
+
}
134
+
135
+
return notFound(res);
136
+
} catch (error) {
137
+
sendJson(res, 500, {
138
+
error: error && error.message ? error.message : 'internal error',
139
+
});
140
+
}
141
+
});
142
+
143
+
server.listen(port, host, () => {
144
+
const address = server.address();
145
+
const origin = `http://${host}:${address.port}`;
146
+
fs.writeFileSync(readyFile, JSON.stringify({ origin }) + '\n', 'utf8');
147
+
});
148
+
149
+
const shutdown = () => {
150
+
server.close(() => process.exit(0));
151
+
};
152
+
153
+
process.on('SIGINT', shutdown);
154
+
process.on('SIGTERM', shutdown);
+67
tools/differential/reference-pds-runner.cjs
+67
tools/differential/reference-pds-runner.cjs
···
1
+
#!/usr/bin/env node
2
+
3
+
const fs = require('node:fs');
4
+
const path = require('node:path');
5
+
const {
6
+
PDS,
7
+
readEnv,
8
+
envToCfg,
9
+
envToSecrets,
10
+
} = require(path.join(
11
+
__dirname,
12
+
'..',
13
+
'..',
14
+
'.tools',
15
+
'reference-runtime',
16
+
'node_modules',
17
+
'@atproto',
18
+
'pds',
19
+
));
20
+
21
+
const readyFile = process.env.PERLDS_READY_FILE;
22
+
23
+
if (!readyFile) {
24
+
console.error('PERLDS_READY_FILE is required');
25
+
process.exit(1);
26
+
}
27
+
28
+
let app;
29
+
30
+
const boot = async () => {
31
+
fs.mkdirSync(process.env.PDS_DATA_DIRECTORY, { recursive: true });
32
+
fs.mkdirSync(process.env.PDS_BLOBSTORE_DISK_LOCATION, { recursive: true });
33
+
if (process.env.PDS_BLOBSTORE_DISK_TMP_LOCATION) {
34
+
fs.mkdirSync(process.env.PDS_BLOBSTORE_DISK_TMP_LOCATION, { recursive: true });
35
+
}
36
+
37
+
const env = readEnv();
38
+
const cfg = envToCfg(env);
39
+
const secrets = envToSecrets(env);
40
+
41
+
app = await PDS.create(cfg, secrets);
42
+
await app.start();
43
+
44
+
fs.writeFileSync(
45
+
readyFile,
46
+
JSON.stringify({ origin: cfg.service.publicUrl, port: cfg.service.port }) + '\n',
47
+
'utf8',
48
+
);
49
+
};
50
+
51
+
const shutdown = async () => {
52
+
try {
53
+
if (app) {
54
+
await app.destroy();
55
+
}
56
+
} finally {
57
+
process.exit(0);
58
+
}
59
+
};
60
+
61
+
process.on('SIGINT', () => void shutdown());
62
+
process.on('SIGTERM', () => void shutdown());
63
+
64
+
boot().catch((error) => {
65
+
console.error(error && error.stack ? error.stack : error);
66
+
process.exit(1);
67
+
});