5835751495

pending completion

Build # 5835751495

Build Type

push

github

Committed by

lucaslorentz

Commit Message

Add husky and apply some code fixes

Run Details

2502 of 2502 new or added lines in 91 files covered. (100.0%)

2286 of 2792 relevant lines covered (81.88%)

143.14 hits per line

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

84.06

/src/LLL.DurableTask.EFCore.SqlServer/SqlServerOrchestrationDbContextExtensions.cs

﻿using System;
using System.Data;
using System.Linq;
using System.Threading.Tasks;
using LLL.DurableTask.EFCore.Entities;
using Microsoft.EntityFrameworkCore;

namespace LLL.DurableTask.EFCore.SqlServer;

public class SqlServerOrchestrationDbContextExtensions : OrchestrationDbContextExtensions
{
    private const IsolationLevel TransactionIsolationLevel = IsolationLevel.ReadCommitted;

    public override async Task Migrate(OrchestrationDbContext dbContext)
    {
        await dbContext.Database.MigrateAsync();
    }

    public override async Task WithinTransaction(OrchestrationDbContext dbContext, Func<Task> action)
    {
        using var transaction = dbContext.Database.BeginTransaction(TransactionIsolationLevel);
        await action();

        await transaction.CommitAsync();
    }

    public override async Task<Instance> LockInstanceForUpdate(OrchestrationDbContext dbContext, string instanceId)
    {
        return (await dbContext.Instances.FromSqlRaw(@"
                SELECT * FROM Instances WITH (UPDLOCK)
                WHERE InstanceId = {0}
            ", instanceId).ToArrayAsync()).FirstOrDefault();
    }

    public override async Task<Instance> TryLockNextInstanceAsync(
        OrchestrationDbContext dbContext,
        TimeSpan lockTimeout)
    {
        var instance = (await dbContext.Instances.FromSqlRaw(@"
                    SELECT TOP 1 Instances.*
                    FROM OrchestrationMessages WITH (INDEX(IX_OrchestrationMessages_AvailableAt_Queue_InstanceId))
                        INNER JOIN Instances WITH (UPDLOCK, READPAST, INDEX(IX_Instances_InstanceId_LockedUntil))
                            ON OrchestrationMessages.InstanceId = Instances.InstanceId
                    WHERE
                        OrchestrationMessages.AvailableAt <= {0}
                        AND Instances.LockedUntil <= {0}
                ", DateTime.UtcNow).ToArrayAsync()).FirstOrDefault();

        if (instance == null)
            return null;

        instance.LockId = Guid.NewGuid().ToString();
        instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);
        await dbContext.SaveChangesAsync();

        return instance;
    }

    public override async Task<Instance> TryLockNextInstanceAsync(
        OrchestrationDbContext dbContext,
        string[] queues,
        TimeSpan lockTimeout)
    {
        var queuesParams = string.Join(",", queues.Select((_, i) => $"{{{i}}}"));
        var utcNowParam = $"{{{queues.Length}}}";
        var parameters = queues.Cast<object>().Concat(new object[] { DateTime.UtcNow }).ToArray();

        var instance = (await dbContext.Instances.FromSqlRaw($@"
                    SELECT TOP 1 Instances.*
                    FROM OrchestrationMessages WITH (INDEX(IX_OrchestrationMessages_AvailableAt_Queue_InstanceId))
                        INNER JOIN Instances WITH (UPDLOCK, READPAST, INDEX(IX_Instances_InstanceId_LockedUntil))
                            ON OrchestrationMessages.InstanceId = Instances.InstanceId
                    WHERE
                        OrchestrationMessages.AvailableAt <= {utcNowParam}
                        AND OrchestrationMessages.Queue IN ({queuesParams})
                        AND Instances.LockedUntil <= {utcNowParam}
                ", parameters).ToArrayAsync()).FirstOrDefault();

        if (instance == null)
            return null;

        instance.LockId = Guid.NewGuid().ToString();
        instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);
        await dbContext.SaveChangesAsync();

        return instance;
    }

    public override async Task<ActivityMessage> TryLockNextActivityMessageAsync(
        OrchestrationDbContext dbContext,
        TimeSpan lockTimeout)
    {
        var instance = (await dbContext.ActivityMessages.FromSqlRaw(@"
                    SELECT TOP 1 *
                    FROM ActivityMessages WITH (UPDLOCK, READPAST, INDEX(IX_ActivityMessages_LockedUntil_Queue))
                    WHERE LockedUntil <= {0}
                ", DateTime.UtcNow).ToArrayAsync()).FirstOrDefault();

        if (instance == null)
            return null;

        instance.LockId = Guid.NewGuid().ToString();
        instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);
        await dbContext.SaveChangesAsync();

        return instance;
    }

    public override async Task<ActivityMessage> TryLockNextActivityMessageAsync(
        OrchestrationDbContext dbContext,
        string[] queues,
        TimeSpan lockTimeout)
    {
        var queuesParams = string.Join(",", queues.Select((_, i) => $"{{{i}}}"));
        var utcNowParam = $"{{{queues.Length}}}";
        var parameters = queues.Cast<object>().Concat(new object[] { DateTime.UtcNow }).ToArray();

        var instance = (await dbContext.ActivityMessages.FromSqlRaw($@"
                SELECT TOP 1 *
                FROM ActivityMessages
                WITH (UPDLOCK, READPAST, INDEX(IX_ActivityMessages_LockedUntil_Queue))
                WHERE Queue IN ({queuesParams})
                    AND LockedUntil <= {utcNowParam}
            ", parameters).ToArrayAsync()).FirstOrDefault();

        if (instance == null)
            return null;

        instance.LockId = Guid.NewGuid().ToString();
        instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);
        await dbContext.SaveChangesAsync();

        return instance;
    }
}

1	using System;
2	using System.Data;
3	using System.Linq;
4	using System.Threading.Tasks;
5	using LLL.DurableTask.EFCore.Entities;
6	using Microsoft.EntityFrameworkCore;
7
8	namespace LLL.DurableTask.EFCore.SqlServer;
9
10	public class SqlServerOrchestrationDbContextExtensions : OrchestrationDbContextExtensions
11	{
12	private const IsolationLevel TransactionIsolationLevel = IsolationLevel.ReadCommitted;
13
14	public override async Task Migrate(OrchestrationDbContext dbContext)
15	{
16	await dbContext.Database.MigrateAsync();	17✔
17	}
18
19	public override async Task WithinTransaction(OrchestrationDbContext dbContext, Func<Task> action)
20	{
21	using var transaction = dbContext.Database.BeginTransaction(TransactionIsolationLevel);	226✔
22	await action();	226✔
23
24	await transaction.CommitAsync();	226✔
25	}
26
27	public override async Task<Instance> LockInstanceForUpdate(OrchestrationDbContext dbContext, string instanceId)
28	{
29	return (await dbContext.Instances.FromSqlRaw(@"	31✔
30	SELECT * FROM Instances WITH (UPDLOCK)	31✔
31	WHERE InstanceId = {0}	31✔
32	", instanceId).ToArrayAsync()).FirstOrDefault();	31✔
33	}
34
35	public override async Task<Instance> TryLockNextInstanceAsync(
36	OrchestrationDbContext dbContext,
37	TimeSpan lockTimeout)
38	{
39	var instance = (await dbContext.Instances.FromSqlRaw(@"	5✔
40	SELECT TOP 1 Instances.*	5✔
41	FROM OrchestrationMessages WITH (INDEX(IX_OrchestrationMessages_AvailableAt_Queue_InstanceId))	5✔
42	INNER JOIN Instances WITH (UPDLOCK, READPAST, INDEX(IX_Instances_InstanceId_LockedUntil))	5✔
43	ON OrchestrationMessages.InstanceId = Instances.InstanceId	5✔
44	WHERE	5✔
45	OrchestrationMessages.AvailableAt <= {0}	5✔
46	AND Instances.LockedUntil <= {0}	5✔
47	", DateTime.UtcNow).ToArrayAsync()).FirstOrDefault();	5✔
48
49	if (instance == null)	5✔
50	return null;	1✔
51
52	instance.LockId = Guid.NewGuid().ToString();	4✔
53	instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);	4✔
54	await dbContext.SaveChangesAsync();	4✔
55
56	return instance;	4✔
57	}
58
59	public override async Task<Instance> TryLockNextInstanceAsync(
60	OrchestrationDbContext dbContext,
61	string[] queues,
62	TimeSpan lockTimeout)
63	{
64	var queuesParams = string.Join(",", queues.Select((_, i) => $"{{{i}}}"));	745✔
65	var utcNowParam = $"{{{queues.Length}}}";	95✔
66	var parameters = queues.Cast<object>().Concat(new object[] { DateTime.UtcNow }).ToArray();	95✔
67
68	var instance = (await dbContext.Instances.FromSqlRaw($@"	95✔
69	SELECT TOP 1 Instances.*	95✔
70	FROM OrchestrationMessages WITH (INDEX(IX_OrchestrationMessages_AvailableAt_Queue_InstanceId))	95✔
71	INNER JOIN Instances WITH (UPDLOCK, READPAST, INDEX(IX_Instances_InstanceId_LockedUntil))	95✔
72	ON OrchestrationMessages.InstanceId = Instances.InstanceId	95✔
73	WHERE	95✔
74	OrchestrationMessages.AvailableAt <= {utcNowParam}	95✔
75	AND OrchestrationMessages.Queue IN ({queuesParams})	95✔
76	AND Instances.LockedUntil <= {utcNowParam}	95✔
77	", parameters).ToArrayAsync()).FirstOrDefault();	95✔
78
79	if (instance == null)	95✔
80	return null;	65✔
81
82	instance.LockId = Guid.NewGuid().ToString();	30✔
83	instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);	30✔
84	await dbContext.SaveChangesAsync();	30✔
85
86	return instance;	30✔
87	}
88
89	public override async Task<ActivityMessage> TryLockNextActivityMessageAsync(
90	OrchestrationDbContext dbContext,
91	TimeSpan lockTimeout)
92	{
93	var instance = (await dbContext.ActivityMessages.FromSqlRaw(@"	×
94	SELECT TOP 1 *	×
95	FROM ActivityMessages WITH (UPDLOCK, READPAST, INDEX(IX_ActivityMessages_LockedUntil_Queue))	×
96	WHERE LockedUntil <= {0}	×
97	", DateTime.UtcNow).ToArrayAsync()).FirstOrDefault();	×
98
99	if (instance == null)	×
100	return null;	×
101
102	instance.LockId = Guid.NewGuid().ToString();	×
103	instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);	×
104	await dbContext.SaveChangesAsync();	×
105
106	return instance;	×
107	}
108
109	public override async Task<ActivityMessage> TryLockNextActivityMessageAsync(
110	OrchestrationDbContext dbContext,
111	string[] queues,
112	TimeSpan lockTimeout)
113	{
114	var queuesParams = string.Join(",", queues.Select((_, i) => $"{{{i}}}"));	312✔
115	var utcNowParam = $"{{{queues.Length}}}";	78✔
116	var parameters = queues.Cast<object>().Concat(new object[] { DateTime.UtcNow }).ToArray();	78✔
117
118	var instance = (await dbContext.ActivityMessages.FromSqlRaw($@"	78✔
119	SELECT TOP 1 *	78✔
120	FROM ActivityMessages	78✔
121	WITH (UPDLOCK, READPAST, INDEX(IX_ActivityMessages_LockedUntil_Queue))	78✔
122	WHERE Queue IN ({queuesParams})	78✔
123	AND LockedUntil <= {utcNowParam}	78✔
124	", parameters).ToArrayAsync()).FirstOrDefault();	78✔
125
126	if (instance == null)	78✔
127	return null;	60✔
128
129	instance.LockId = Guid.NewGuid().ToString();	18✔
130	instance.LockedUntil = DateTime.UtcNow.Add(lockTimeout);	18✔
131	await dbContext.SaveChangesAsync();	18✔
132
133	return instance;	18✔
134	}
135	}

lucaslorentz / durabletask-extensions / 5835751495

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