10439138780

Committed 18 Aug 2024 08:23AM UTC coverage: 67.144% (+0.06%) from 67.084%

Build # 10439138780

Build Type

push

github

Committed by

mt89vein

Commit Message

feat: outbox with autopartitioning

Run Details

199 of 376 branches covered (52.93%)

Branch coverage included in aggregate %.

256 of 367 new or added lines in 18 files covered. (69.75%)

4 existing lines in 2 files now uncovered.

833 of 1161 relevant lines covered (71.75%)

69.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.74

/src/Sstv.Outbox.Npgsql/StrictOrderingOutboxRepository.cs

using System.Diagnostics.CodeAnalysis;
using Dapper;
using Microsoft.Extensions.Options;
using Npgsql;

namespace Sstv.Outbox.Npgsql;

/// <summary>
/// Implements API for single active worker.
/// </summary>
[SuppressMessage("Security", "CA2100:sql injection check", Justification = "There is no user input in sql")]
public sealed class StrictOrderingOutboxRepository<TOutboxItem> : IOutboxRepository<TOutboxItem>
    where TOutboxItem : class, IOutboxItem
{
    private readonly OutboxOptions _options;
    private NpgsqlConnection? _connection;
    private NpgsqlTransaction? _transaction;

    /// <summary>
    /// Creates new instance of <see cref="StrictOrderingOutboxRepository{TOutboxItem}"/>.
    /// </summary>
    /// <param name="options">Outbox Options.</param>
    public StrictOrderingOutboxRepository(IOptionsMonitor<OutboxOptions> options)
    {
        ArgumentNullException.ThrowIfNull(options);

        _options = options.Get(typeof(TOutboxItem).Name);
    }

    /// <summary>
    /// Lock, fetch and return outbox items.
    /// </summary>
    /// <param name="ct">Token for cancel operation.</param>
    /// <returns>OutboxItems.</returns>
    public async Task<IEnumerable<TOutboxItem>> LockAndReturnItemsBatchAsync(CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();

        _connection = await _options.GetNpgsqlDataSource().OpenConnectionAsync(ct);
        _transaction = await _connection.BeginTransactionAsync(ct);

        var m = _options.GetDbMapping();

        var filter = _options.PartitionSettings.Enabled
            ? $"WHERE {m.Status} <> {(int)OutboxItemStatus.Completed}"
            : string.Empty;

        var order = _options.GetPriorityFeature()
            .Enabled
            ? $"ORDER BY {m.Priority} DESC, {m.Id} ASC"
            : $"ORDER BY {m.Id} ASC";

        var sql = $"""
                   SELECT * FROM {m.QualifiedTableName}
                   {filter}
                   {order}
                   LIMIT {_options.OutboxItemsLimit}
                   FOR UPDATE NOWAIT;
                   """;

        try
        {
            return await _connection.QueryAsync<TOutboxItem>(sql, transaction: _transaction);
        }
        catch (PostgresException e) when (e.SqlState == PostgresErrorCodes.LockNotAvailable)
        {
            return Array.Empty<TOutboxItem>();
        }
    }

    /// <summary>
    /// Saves results.
    /// </summary>
    /// <param name="completed">Completed outbox items.</param>
    /// <param name="retried">Retries outbox items.</param>
    /// <param name="ct">Token for cancel operation.</param>
    public async Task SaveAsync(
        IReadOnlyCollection<TOutboxItem> completed,
        IReadOnlyCollection<TOutboxItem> retried,
        CancellationToken ct
    )
    {
        ArgumentNullException.ThrowIfNull(completed);
        ArgumentNullException.ThrowIfNull(retried);

        if (_transaction is null)
        {
            throw new InvalidOperationException("Transaction was null");
        }

        if (retried.Count != 0)
        {
            throw new NotSupportedException("Retry not supported for strict ordering worker");
        }

        var m = _options.GetDbMapping();

        if (!_options.PartitionSettings.Enabled)
        {
            const string ids = "ids";
            var sql = $"""
                       DELETE FROM {m.QualifiedTableName}
                       WHERE {m.Id} in (select * from unnest(@{ids}));
                       """;

            await using var cmd = _connection!.CreateCommand();
            cmd.CommandText = sql;
            cmd.Parameters.Add(new NpgsqlParameter<Guid[]>(ids, completed
                .Select(o => o.Id)
                .ToArray()));

            await cmd.ExecuteNonQueryAsync(ct);
        }
        else
        {
            const string ids = "ids";
            var sql = $"""
                       UPDATE {m.QualifiedTableName}
                       SET "{m.Status}" = {(int)OutboxItemStatus.Completed}
                       WHERE {m.Id} in (select * from unnest(@{ids}));
                       """;

            await using var cmd = _connection!.CreateCommand();
            cmd.CommandText = sql;
            cmd.Parameters.Add(new NpgsqlParameter<Guid[]>(ids, completed
                .Select(o => o.Id)
                .ToArray()));
            await cmd.ExecuteNonQueryAsync(ct);
        }

        await _transaction.CommitAsync(ct);
    }

    /// <summary>
    /// Cleans resources.
    /// </summary>
    public void Dispose()
    {
        _connection?.Dispose();
        _transaction?.Dispose();
    }

    /// <summary>
    /// Cleans resources.
    /// </summary>
    public async ValueTask DisposeAsync()
    {
        if (_connection != null)
        {
            await _connection.DisposeAsync();
        }

        if (_transaction != null)
        {
            await _transaction.DisposeAsync();
        }
    }
}

1	using System.Diagnostics.CodeAnalysis;
2	using Dapper;
3	using Microsoft.Extensions.Options;
4	using Npgsql;
5
6	namespace Sstv.Outbox.Npgsql;
7
8	/// <summary>
9	/// Implements API for single active worker.
10	/// </summary>
11	[SuppressMessage("Security", "CA2100:sql injection check", Justification = "There is no user input in sql")]
12	public sealed class StrictOrderingOutboxRepository<TOutboxItem> : IOutboxRepository<TOutboxItem>
13	where TOutboxItem : class, IOutboxItem
14	{
15	private readonly OutboxOptions _options;
16	private NpgsqlConnection? _connection;
17	private NpgsqlTransaction? _transaction;
18
19	/// <summary>
20	/// Creates new instance of <see cref="StrictOrderingOutboxRepository{TOutboxItem}"/>.
21	/// </summary>
22	/// <param name="options">Outbox Options.</param>
23	public StrictOrderingOutboxRepository(IOptionsMonitor<OutboxOptions> options)	12✔
24	{
25	ArgumentNullException.ThrowIfNull(options);	12✔
26
27	_options = options.Get(typeof(TOutboxItem).Name);	12✔
28	}	12✔
29
30	/// <summary>
31	/// Lock, fetch and return outbox items.
32	/// </summary>
33	/// <param name="ct">Token for cancel operation.</param>
34	/// <returns>OutboxItems.</returns>
35	public async Task<IEnumerable<TOutboxItem>> LockAndReturnItemsBatchAsync(CancellationToken ct = default)
36	{
37	ct.ThrowIfCancellationRequested();	12✔
38
39	_connection = await _options.GetNpgsqlDataSource().OpenConnectionAsync(ct);	12✔
40	_transaction = await _connection.BeginTransactionAsync(ct);	12✔
41
42	var m = _options.GetDbMapping();	12✔
43
44	var filter = _options.PartitionSettings.Enabled	12!
45	? $"WHERE {m.Status} <> {(int)OutboxItemStatus.Completed}"	12✔
46	: string.Empty;	12✔
47
48	var order = _options.GetPriorityFeature()	12!
49	.Enabled	12✔
50	? $"ORDER BY {m.Priority} DESC, {m.Id} ASC"	12✔
51	: $"ORDER BY {m.Id} ASC";	12✔
52
53	var sql = $"""	12✔
54	SELECT * FROM {m.QualifiedTableName}	12✔
55	{filter}	12✔
56	{order}	12✔
57	LIMIT {_options.OutboxItemsLimit}	12✔
58	FOR UPDATE NOWAIT;	12✔
59	""";	12✔
60
61	try
62	{
63	return await _connection.QueryAsync<TOutboxItem>(sql, transaction: _transaction);	12✔
64	}
65	catch (PostgresException e) when (e.SqlState == PostgresErrorCodes.LockNotAvailable)	9✔
66	{
67	return Array.Empty<TOutboxItem>();	9✔
68	}
69	}	12✔
70
71	/// <summary>
72	/// Saves results.
73	/// </summary>
74	/// <param name="completed">Completed outbox items.</param>
75	/// <param name="retried">Retries outbox items.</param>
76	/// <param name="ct">Token for cancel operation.</param>
77	public async Task SaveAsync(
78	IReadOnlyCollection<TOutboxItem> completed,
79	IReadOnlyCollection<TOutboxItem> retried,
80	CancellationToken ct
81	)
82	{
83	ArgumentNullException.ThrowIfNull(completed);	3✔
84	ArgumentNullException.ThrowIfNull(retried);	3✔
85
86	if (_transaction is null)	3!
87	{
88	throw new InvalidOperationException("Transaction was null");	×
89	}
90
91	if (retried.Count != 0)	3!
92	{
93	throw new NotSupportedException("Retry not supported for strict ordering worker");	×
94	}
95
96	var m = _options.GetDbMapping();	3✔
97
98	if (!_options.PartitionSettings.Enabled)	3!
99	{
100	const string ids = "ids";
101	var sql = $"""	3✔
102	DELETE FROM {m.QualifiedTableName}	3✔
103	WHERE {m.Id} in (select * from unnest(@{ids}));	3✔
104	""";	3✔
105
106	await using var cmd = _connection!.CreateCommand();	3✔
107	cmd.CommandText = sql;	3✔
108	cmd.Parameters.Add(new NpgsqlParameter<Guid[]>(ids, completed	3✔
109	.Select(o => o.Id)	22✔
110	.ToArray()));	3✔
111
112	await cmd.ExecuteNonQueryAsync(ct);	3✔
113	}	3✔
114	else
115	{
116	const string ids = "ids";
NEW 117	var sql = $"""	×
NEW 118	UPDATE {m.QualifiedTableName}	×
NEW 119	SET "{m.Status}" = {(int)OutboxItemStatus.Completed}	×
NEW 120	WHERE {m.Id} in (select * from unnest(@{ids}));	×
NEW 121	""";	×
122
NEW 123	await using var cmd = _connection!.CreateCommand();	×
NEW 124	cmd.CommandText = sql;	×
NEW 125	cmd.Parameters.Add(new NpgsqlParameter<Guid[]>(ids, completed	×
NEW 126	.Select(o => o.Id)	×
NEW 127	.ToArray()));	×
NEW 128	await cmd.ExecuteNonQueryAsync(ct);	×
NEW 129	}	×
130
131	await _transaction.CommitAsync(ct);	3✔
132	}	3✔
133
134	/// <summary>
135	/// Cleans resources.
136	/// </summary>
137	public void Dispose()
138	{
139	_connection?.Dispose();	×
140	_transaction?.Dispose();	×
141	}	×
142
143	/// <summary>
144	/// Cleans resources.
145	/// </summary>
146	public async ValueTask DisposeAsync()
147	{
148	if (_connection != null)	24✔
149	{
150	await _connection.DisposeAsync();	24✔
151	}
152
153	if (_transaction != null)	24✔
154	{
155	await _transaction.DisposeAsync();	24✔
156	}
157	}	24✔
158	}

mt89vein / Sstv.Outbox / 10439138780

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous