Skip to content

Commit 68e21e8

Browse files
ChadSikorraChadSikorra
committed
Make replica local password state changes atomic with a lock. Add sequencing markers so we can tell which has been forwarded.
1 parent d427ab3 commit 68e21e8

14 files changed

Lines changed: 464 additions & 55 deletions

File tree

‎resources/schema/mysql/baseline.sql‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,9 @@ CREATE TABLE IF NOT EXISTS ldap_change_journal_seq (
4545
INSERT IGNORE INTO ldap_change_journal_seq (id, seq) VALUES (1, 0);
4646

4747
CREATE TABLE IF NOT EXISTS ldap_replica_pwpolicy_state (
48-
lc_dn VARCHAR(768) NOT NULL,
49-
state JSON NOT NULL,
48+
lc_dn VARCHAR(768) NOT NULL,
49+
state JSON NOT NULL,
50+
seq BIGINT NOT NULL DEFAULT 0,
51+
forwarded_seq BIGINT NOT NULL DEFAULT 0,
5052
PRIMARY KEY (lc_dn)
5153
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

‎resources/schema/sqlite/baseline.sql‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,8 @@ CREATE TABLE IF NOT EXISTS ldap_change_journal_seq (
4747
INSERT OR IGNORE INTO ldap_change_journal_seq (id, seq) VALUES (1, 0);
4848

4949
CREATE TABLE IF NOT EXISTS ldap_replica_pwpolicy_state (
50-
lc_dn TEXT NOT NULL PRIMARY KEY,
51-
state TEXT NOT NULL
50+
lc_dn TEXT NOT NULL PRIMARY KEY,
51+
state TEXT NOT NULL,
52+
seq INTEGER NOT NULL DEFAULT 0,
53+
forwarded_seq INTEGER NOT NULL DEFAULT 0
5254
);

‎src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/Dialect/MysqlDialect.php‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,16 +34,17 @@ public function createFilterTranslator(?SubstringIndexInterface $substringIndex)
3434
return new MysqlFilterTranslator($substringIndex);
3535
}
3636

37-
public function lockEntryForWrite(
37+
public function lockRowForWrite(
3838
PDO $pdo,
39+
string $table,
3940
string $lcDn,
4041
): void {
4142
$statement = $pdo->prepare(<<<SQL
4243
SELECT lc_dn
43-
FROM entries
44+
FROM $table
4445
WHERE lc_dn = ?
4546
FOR UPDATE
46-
SQL);
47+
SQL);
4748
$statement->execute([$lcDn]);
4849
}
4950

‎src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/Dialect/PdoDialectTrait.php‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,10 +38,11 @@ public function rollBack(PDO $pdo): void
3838
}
3939

4040
/**
41-
* Default no-op: not all implementations need this.
41+
* Default no-op: SQLite already holds the write lock from `BEGIN IMMEDIATE`, so no per-row lock is needed.
4242
*/
43-
public function lockEntryForWrite(
43+
public function lockRowForWrite(
4444
PDO $pdo,
45+
string $table,
4546
string $lcDn,
4647
): void {}
4748

‎src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/Dialect/PdoEntryDialectInterface.php‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,10 +33,13 @@ public function beginTransaction(PDO $pdo): void;
3333
public function commit(PDO $pdo): void;
3434

3535
/**
36-
* Take an exclusive lock on the entry row within the current transaction so concurrent writers to it serialize.
36+
* Take an exclusive lock on the lc_dn row of $table within the current transaction so concurrent writers serialize.
37+
*
38+
* @param string $table A fixed internal table identifier, never client input, so implementations may interpolate it.
3739
*/
38-
public function lockEntryForWrite(
40+
public function lockRowForWrite(
3941
PDO $pdo,
42+
string $table,
4043
string $lcDn,
4144
): void;
4245

‎src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/Pdo/PdoColumnCastTrait.php‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
/**
6+
* This file is part of the FreeDSx LDAP package.
7+
*
8+
* (c) Chad Sikorra <Chad.Sikorra@gmail.com>
9+
*
10+
* For the full copyright and license information, please view the LICENSE
11+
* file that was distributed with this source code.
12+
*/
13+
14+
namespace FreeDSx\Ldap\Server\Backend\Storage\Adapter\Pdo;
15+
16+
use function is_numeric;
17+
use function is_string;
18+
19+
/**
20+
* Narrows loosely-typed PDO fetched column values to their expected scalar type.
21+
*
22+
* @author Chad Sikorra <Chad.Sikorra@gmail.com>
23+
*/
24+
trait PdoColumnCastTrait
25+
{
26+
private function stringColumn(mixed $value): string
27+
{
28+
return is_string($value)
29+
? $value
30+
: '';
31+
}
32+
33+
private function intColumn(mixed $value): int
34+
{
35+
return is_numeric($value)
36+
? (int) $value
37+
: 0;
38+
}
39+
}

‎src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/PdoReplicaPasswordStateStore.php‎

Lines changed: 100 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -14,16 +14,20 @@
1414
namespace FreeDSx\Ldap\Server\Backend\Storage\Adapter;
1515

1616
use FreeDSx\Ldap\Entry\Dn;
17+
use FreeDSx\Ldap\Server\Backend\Storage\Adapter\Dialect\PdoEntryDialectInterface;
18+
use FreeDSx\Ldap\Server\Backend\Storage\Adapter\Pdo\PdoColumnCastTrait;
1719
use FreeDSx\Ldap\Server\Backend\Storage\Adapter\Pdo\PdoTransactor;
1820
use FreeDSx\Ldap\Server\Backend\Storage\Exception\StorageIoException;
1921
use FreeDSx\Ldap\Server\PasswordPolicy\Decision\OperationalChanges;
22+
use FreeDSx\Ldap\Server\PasswordPolicy\Replica\ReplicaForwardState;
2023
use FreeDSx\Ldap\Server\PasswordPolicy\Replica\ReplicaPasswordState;
2124
use FreeDSx\Ldap\Server\PasswordPolicy\Replica\ReplicaPasswordStateStoreInterface;
25+
use PDO;
2226

2327
use function is_array;
24-
use function is_string;
2528
use function json_decode;
2629
use function json_encode;
30+
use function max;
2731

2832
/**
2933
* Replica-local password-policy state persisted as a JSON row per subject, sharing a PdoStorage connection.
@@ -32,21 +36,18 @@
3236
*/
3337
final readonly class PdoReplicaPasswordStateStore implements ReplicaPasswordStateStoreInterface
3438
{
39+
use PdoColumnCastTrait;
40+
3541
private const TABLE = 'ldap_replica_pwpolicy_state';
3642

37-
public function __construct(private PdoTransactor $transactor) {}
43+
public function __construct(
44+
private PdoTransactor $transactor,
45+
private PdoEntryDialectInterface $dialect,
46+
) {}
3847

3948
public function load(Dn $dn): ReplicaPasswordState
4049
{
41-
$statement = $this->transactor
42-
->pdo()
43-
->prepare('SELECT state FROM ' . self::TABLE . ' WHERE lc_dn = ?');
44-
$statement->execute([$this->key($dn)]);
45-
$state = $statement->fetchColumn();
46-
47-
return is_string($state)
48-
? $this->decode($state)
49-
: ReplicaPasswordState::empty();
50+
return $this->loadRecord($dn)->state;
5051
}
5152

5253
/**
@@ -57,35 +58,104 @@ public function atomicMutate(
5758
callable $merge,
5859
): void {
5960
$this->transactor->atomic(function () use ($dn, $merge): void {
60-
$key = $this->key($dn);
61-
$current = $this->load($dn);
62-
$changes = $merge($current);
63-
61+
$this->dialect->lockRowForWrite(
62+
$this->transactor->pdo(),
63+
self::TABLE,
64+
$this->key($dn),
65+
);
66+
67+
$record = $this->loadRecord($dn);
68+
$changes = $merge($record->state);
6469
if ($changes->isEmpty()) {
6570
return;
6671
}
6772

68-
$next = $current->withChanges($changes);
69-
70-
$this->transactor
71-
->pdo()
72-
->prepare('DELETE FROM ' . self::TABLE . ' WHERE lc_dn = ?')
73-
->execute([$key]);
74-
75-
if ($next->isEmpty()) {
73+
$next = $record->state->withChanges($changes);
74+
if ($record->state->equals($next)) {
7675
return;
7776
}
7877

79-
$this->transactor
80-
->pdo()
81-
->prepare('INSERT INTO ' . self::TABLE . ' (lc_dn, state) VALUES (?, ?)')
82-
->execute([
83-
$key,
84-
$this->encode($next),
85-
]);
78+
$this->upsert($record->applied($next));
8679
});
8780
}
8881

82+
public function listUnforwarded(int $limit = 100): array
83+
{
84+
$statement = $this->transactor
85+
->pdo()
86+
->prepare(
87+
'SELECT lc_dn, state, seq, forwarded_seq FROM ' . self::TABLE
88+
. ' WHERE seq > forwarded_seq ORDER BY seq ASC LIMIT ' . max(0, $limit),
89+
);
90+
$statement->execute();
91+
92+
$pending = [];
93+
/** @var array<string, mixed> $row */
94+
foreach ($statement->fetchAll(PDO::FETCH_ASSOC) as $row) {
95+
$pending[] = new ReplicaForwardState(
96+
new Dn($this->stringColumn($row['lc_dn'])),
97+
$this->decode($this->stringColumn($row['state'])),
98+
$this->intColumn($row['seq']),
99+
$this->intColumn($row['forwarded_seq']),
100+
);
101+
}
102+
103+
return $pending;
104+
}
105+
106+
public function markForwarded(
107+
Dn $dn,
108+
int $sequence,
109+
): void {
110+
$this->transactor
111+
->pdo()
112+
->prepare(
113+
'UPDATE ' . self::TABLE
114+
. ' SET forwarded_seq = ? WHERE lc_dn = ? AND forwarded_seq < ? AND seq >= ?',
115+
)
116+
->execute([
117+
$sequence,
118+
$this->key($dn),
119+
$sequence,
120+
$sequence,
121+
]);
122+
}
123+
124+
private function loadRecord(Dn $dn): ReplicaForwardState
125+
{
126+
$statement = $this->transactor
127+
->pdo()
128+
->prepare('SELECT state, seq, forwarded_seq FROM ' . self::TABLE . ' WHERE lc_dn = ?');
129+
$statement->execute([$this->key($dn)]);
130+
$row = $statement->fetch(PDO::FETCH_ASSOC);
131+
if (!is_array($row)) {
132+
return ReplicaForwardState::initial($dn);
133+
}
134+
135+
return new ReplicaForwardState(
136+
$dn,
137+
$this->decode($this->stringColumn($row['state'])),
138+
$this->intColumn($row['seq']),
139+
$this->intColumn($row['forwarded_seq']),
140+
);
141+
}
142+
143+
private function upsert(ReplicaForwardState $record): void
144+
{
145+
$key = $this->key($record->dn);
146+
$pdo = $this->transactor->pdo();
147+
148+
$pdo->prepare('DELETE FROM ' . self::TABLE . ' WHERE lc_dn = ?')
149+
->execute([$key]);
150+
$pdo->prepare('INSERT INTO ' . self::TABLE . ' (lc_dn, state, seq, forwarded_seq) VALUES (?, ?, ?, ?)')
151+
->execute([
152+
$key,
153+
$this->encode($record->state),
154+
$record->sequence,
155+
$record->forwarded,
156+
]);
157+
}
158+
89159
private function key(Dn $dn): string
90160
{
91161
return $dn->normalize()->toString();

‎src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/PdoStorage.php‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -293,15 +293,19 @@ public function atomic(callable $operation): void
293293

294294
public function lockForWrite(Dn $dn): void
295295
{
296-
$this->dialect->lockEntryForWrite(
296+
$this->dialect->lockRowForWrite(
297297
$this->transactor->pdo(),
298+
'entries',
298299
$dn->normalize()->toString(),
299300
);
300301
}
301302

302303
public function replicaPasswordStateStore(): ReplicaPasswordStateStoreInterface
303304
{
304-
return new PdoReplicaPasswordStateStore($this->transactor);
305+
return new PdoReplicaPasswordStateStore(
306+
$this->transactor,
307+
$this->dialect,
308+
);
305309
}
306310

307311
protected function buildJournal(ChangeJournalConfig $config): ChangeJournalInterface

‎src/FreeDSx/Ldap/Server/PasswordPolicy/Replica/InMemoryReplicaPasswordStateStore.php‎

Lines changed: 43 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -16,22 +16,23 @@
1616
use FreeDSx\Ldap\Entry\Dn;
1717
use FreeDSx\Ldap\Server\PasswordPolicy\Decision\OperationalChanges;
1818

19+
use function count;
20+
1921
/**
20-
* Holds replica-observed password-policy state in memory.
22+
* Holds replica-observed password-policy state in memory, with a per-subject forward watermark.
2123
*
2224
* @author Chad Sikorra <Chad.Sikorra@gmail.com>
2325
*/
2426
final class InMemoryReplicaPasswordStateStore implements ReplicaPasswordStateStoreInterface
2527
{
2628
/**
27-
* @var array<string, ReplicaPasswordState>
29+
* @var array<string, ReplicaForwardState>
2830
*/
29-
private array $states = [];
31+
private array $records = [];
3032

3133
public function load(Dn $dn): ReplicaPasswordState
3234
{
33-
return $this->states[$this->key($dn)]
34-
?? ReplicaPasswordState::empty();
35+
return ($this->records[$this->key($dn)] ?? ReplicaForwardState::initial($dn))->state;
3536
}
3637

3738
/**
@@ -42,20 +43,51 @@ public function atomicMutate(
4243
callable $merge,
4344
): void {
4445
$key = $this->key($dn);
45-
$current = $this->states[$key] ?? ReplicaPasswordState::empty();
46-
$changes = $merge($current);
46+
$record = $this->records[$key] ?? ReplicaForwardState::initial($dn);
47+
$changes = $merge($record->state);
48+
4749
if ($changes->isEmpty()) {
4850
return;
4951
}
5052

51-
$state = $current->withChanges($changes);
52-
if ($state->isEmpty()) {
53-
unset($this->states[$key]);
53+
$next = $record->state->withChanges($changes);
54+
if ($record->state->equals($next)) {
55+
return;
56+
}
57+
58+
$this->records[$key] = $record->applied($next);
59+
}
60+
61+
public function listUnforwarded(int $limit = 100): array
62+
{
63+
$pending = [];
64+
65+
foreach ($this->records as $record) {
66+
if (!$record->isPending()) {
67+
continue;
68+
}
69+
70+
$pending[] = $record;
71+
if (count($pending) >= $limit) {
72+
break;
73+
}
74+
}
75+
76+
return $pending;
77+
}
78+
79+
public function markForwarded(
80+
Dn $dn,
81+
int $sequence,
82+
): void {
83+
$key = $this->key($dn);
84+
$record = $this->records[$key] ?? null;
5485

86+
if ($record === null || !$record->canAdvanceTo($sequence)) {
5587
return;
5688
}
5789

58-
$this->states[$key] = $state;
90+
$this->records[$key] = $record->advancedTo($sequence);
5991
}
6092

6193
private function key(Dn $dn): string

0 commit comments

Comments
 (0)