Is this table-backed-queue implementation safe? We've a system where kinds of orders should be processed in batches, by advancing them from Status 0 to Status 1. There's a max number of a particular kind of order we can process in any one go, and it depends on the kind of the order, so we might be able to process 10 OrderKind 3's but only 7 OrderKind 5's.

Waiting to go in the orders table might be 20 of OrderKind 3, and 70 of OrderKind 5 etc. So in any one batching operation we only want to pick up max 10 of the OK 3's plus max 7 of the OK 5's, giving a batch of 17 rows total.

Here's the stored procedure that finds the oldest (according to when the order was approved) X orders, updates them to Status 1 and returns them to the front end app.

ALTER   procedure [ProcessPendingOrders]
as  
begin transaction

declare @Processing table
(
    Id int,
    CustomerId uniqueidentifier,
    OrderControlId int,
    Destination nvarchar(256),
    Quantity int
);

update  Orders
set     StatusID = 1 --Being processed
output inserted.Id,
       inserted.CustomerID,
       inserted.OrderControlId,
       inserted.Destination,
       inserted.Quantity
into   @Processing
from    Orders o
where   o.Id in (   select  topN.Id
                    from    OrderControl oc
                    cross apply (   select  top(oc.OrdersPerBatch)
                                            oo.Id   
                                    from    Orders o2
                                    join    OrderControl oc2 on o2.OrderControlId = oc2.Id
                                    where   o2.StatusID = 0       --waiting
                                    and     oc2.ProcessOrders = 1 
                                    and     oc2.ID = oc.ID
                                    order by o2.ApprovalDate asc
                                ) topN
                )

if @@error &lt;&gt; 0
 begin
    rollback
    raiserror ('Error processing orders', 16, 1)
    return
 end

 select * from @Processing

commit


My question is; what would happen if two external processes were to execute this stored procedure concurrently? Is it possible that the same order rows would be given to each process? I've a concern it will because of the extra legwork doing a top N per kind of order. Is there anything that can be done to apply locks such that two different processes can't get the same set of rows out of this procedure?

Would the same problem afflict the earlier iteration of this code that didn't have top N?  It was literally just a single UPDATE orders SET status = 1 OUTPUT blah INTO @abc WHERE status = 0 query.
Check out the locking that occurs in your stored procedure - the following is a simple example, you can adapt it to test your query and others.  In case it needs to be clear, go to a quiet server for this, like a development enivironment or better yet, a local install.  The last thing you want to be doing is holding transactions open for giggles in a production environment.

USE tempdb;
GO

IF ( OBJECT_ID( 'dbo.TestTable', 'U' ) IS NULL )
BEGIN
    --DROP TABLE dbo.TestTable;
    CREATE TABLE dbo.TestTable
    (
        Foo                 INTEGER,
        BAR                 INTEGER
    );

    INSERT  INTO dbo.TestTable( Foo, Bar )
    SELECT  TOP 10 1, 1
    FROM    sys.objects;
END;
GO

IF ( OBJECT_ID( 'dbo.TestProcess', 'P' ) IS NULL )
BEGIN
    --DROP PROCEDURE dbo.TestProcess;
    EXEC( 'CREATE PROCEDURE dbo.TestProcess AS SET NOCOUNT OFF;' );
END;
GO

ALTER PROCEDURE dbo.TestProcess
AS BEGIN
    SET NOCOUNT ON;

    --SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
    BEGIN TRANSACTION;

    SELECT  isolation_level = CASE transaction_isolation_level
                WHEN 0 THEN 'Unspecified'
                WHEN 1 THEN 'Read Uncommitted'
                WHEN 2 THEN 'Read Committed'
                WHEN 3 THEN 'Repeatable'
                WHEN 4 THEN 'Serializable'
                WHEN 5 THEN 'Snapshot'
                ELSE 'UNKNOWN ( ' + CONVERT( NVARCHAR( 8 ), 
                    transaction_isolation_level ) + ' )'
            END
    FROM    sys.dm_exec_requests
    WHERE   session_id = @@SPID;

    SELECT  ObjectName = OBJECT_NAME( resource_associated_entity_id ), 
            request_session_id, resource_type, 
            request_mode, request_type, request_status
    FROM    sys.dm_tran_locks
    WHERE   request_session_id = @@SPID
        AND resource_type = 'OBJECT';

    UPDATE  dbo.TestTable
        SET Foo = 1
    OUTPUT  INSERTED.Foo, INSERTED.Bar
    WHERE   Bar = 1;

    SELECT  ObjectName = OBJECT_NAME( resource_associated_entity_id ), 
            request_session_id, resource_type, 
            request_mode, request_type, request_status
    FROM    sys.dm_tran_locks
    WHERE   request_session_id = @@SPID
        AND resource_type = 'OBJECT';

    IF ( @@ERROR &lt;&gt; 0 )
    BEGIN
        ROLLBACK TRANSACTION;
        RAISERROR( 'Rolled Back Update', 16, 1 ) WITH NOWAIT;
    END ELSE BEGIN
        COMMIT TRANSACTION;
    END;

    SELECT  ObjectName = OBJECT_NAME( resource_associated_entity_id ), 
            request_session_id, resource_type, 
            request_mode, request_type, request_status
    FROM    sys.dm_tran_locks
    WHERE   request_session_id = @@SPID
        AND resource_type = 'OBJECT';

    SET NOCOUNT OFF;

    SELECT  isolation_level = CASE transaction_isolation_level
                WHEN 0 THEN 'Unspecified'
                WHEN 1 THEN 'Read Uncommitted'
                WHEN 2 THEN 'Read Committed'
                WHEN 3 THEN 'Repeatable'
                WHEN 4 THEN 'Serializable'
                WHEN 5 THEN 'Snapshot'
                ELSE 'UNKNOWN ( ' + CONVERT( NVARCHAR( 8 ), 
                    transaction_isolation_level ) + ' )'
            END
    FROM    sys.dm_exec_requests
    WHERE   session_id = @@SPID;
END;
GO


With that set up, open a new session, which opens a transaction and runs the "guts" of the procedure:

USE tempdb;
GO

BEGIN TRANSACTION;

UPDATE  dbo.TestTable
    SET Foo = 1
WHERE   Bar = 1;

--COMMIT TRANSACTION;


Open another session and do the same thing, closing the transaction this time:

USE tempdb;
GO

BEGIN TRANSACTION;

UPDATE  dbo.TestTable
    SET Foo = 1
WHERE   Bar = 1;

COMMIT TRANSACTION;


This example will hang, as update locks are being held on the test table.  In yet another session, run the following:

USE tempdb;
GO

SELECT  ObjectName = OBJECT_NAME( resource_associated_entity_id ), 
        request_session_id, resource_type, request_mode, 
        request_type, request_status
FROM    sys.dm_tran_locks
WHERE   resource_database_id = DB_ID()
    AND resource_type = 'OBJECT';

SELECT  session_id, wait_type, resource_description
FROM    sys.dm_os_waiting_tasks
WHERE   session_id IN ( SELECT  request_session_id  
                        FROM    sys.dm_tran_locks
                        WHERE   resource_database_id = DB_ID()
                            AND resource_type = 'OBJECT' );


Here you will see that when the "guts" are run, transaction locks are granted to both the uncommitted session and the hanging session, but the hanging session is waiting for the modify update lock to be granted ( LCK_M_U ) while it waits for the original session to finish.  Go back to the first session and commit the transaction, then go ahead and run the procedure.

USE tempdb;
GO

EXECUTE dbo.TestProcess;


You should see that same intent-exclusive ( IX ) lock having been granted while the procedure runs.  This will lead to the same scenario as outlined above, where further transactions against the affected objects will "hang" while the active transactions complete / the procedure runs.

This is all using the default SET TRANSACTION ISOLATION LEVEL READ COMMITTED, well, transaction isolation level.  With the TOPs and CROSS APPLYs, it's potentially possible that a READ UNCOMMITTED or even a READ COMMITTED isolation level would allow concurrent transactions to effectively SELECT the keys about to be operated on before granting the LCK_M_Us, but in a single statement such as in your procedure, I would be somewhat surprised if that were the case.  If so, then a SET TRANSACTION ISOLATION LEVEL SERIALIZABLE; may be the best option to ensure asynchronous access to the objects.