Transaction Log Filling The Disk

I am having an issue with the transaction growing uncontrolled and filling up the disk. I suspect that transactions are structured incorrectly between the web application that is monitoring a queue and the SQL that is executing the WAITFOR RECEIVE. This method is receiving large binary objects, so thats the reason for the arguments to the reader. Also, even though its not the suggested way, we commit everytime through to prevent the queue from disabling (which it was doing when we would ROLLBACK - we don't really care if the message is bad, we just want to log it and wait for the next one).

The basic structure is this, which is executed on a separate thread. Am I missing something that could be causing transactions to get into a state where the log grows uncontrollably? Is there a problem with the loop? Should I be doing a ROLLBACK when there is nothing to receive (this is a low volume queue, so it may not receive a message for a few minutes or more)? If so, where should I be doing this?


private void MonitorUpdateQueue()

{

const string _sqlEndDialog = @"END CONVERSATION @conversationHandle";

const string _errorMessageType = @"http://schemas.microsoft.com/SQL/ServiceBroker/Error";

const string _endDialogMessageType = @"http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog";

using (_ssbConnection)

{

if (_ssbConnection == null) _ssbConnection = new SqlConnection();

using (_monitorQueueCommand =

new SqlCommand("asp_ProcessMessage", _ssbConnection))

{

_monitorQueueCommand.CommandType = CommandType.StoredProcedure;

_monitorQueueCommand.CommandTimeout = 0;

while (true)

{

SqlDataReader reader = null;

Guid conversationHandle = Guid.Empty;

Byte[] payloadBytes = { 0 };

try

{

if (_ssbConnection.State == ConnectionState.Closed)

{

_ssbConnection.ConnectionString = _ssbConnectionString;

_ssbConnection.Open();

}

Byte[] latestFullUpdate = null;

_monitorQueueCommand.Transaction = _ssbConnection.BeginTransaction(IsolationLevel.Serializable);

if (_monitorQueueCommand.Transaction != null)

{

reader = _monitorQueueCommand.ExecuteReader(CommandBehavior.SequentialAccess);

if (reader != null)

{

XmlDocument updateMessage = new XmlDocument();

while (reader.Read())

{

conversationHandle = reader.GetGuid(0);

string messageType = reader.GetString(1);

payloadBytes = reader.GetSqlBytes(2).Value;

if (messageType == _endDialogMessageType || messageType == _errorMessageType)

{

SqlCommand cmdEndDialog = new SqlCommand(_sqlEndDialog, _ssbConnection, _monitorQueueCommand.Transaction);

cmdEndDialog.Parameters.AddWithValue("@conversationHandle",

conversationHandle);

cmdEndDialog.ExecuteNonQuery();

}

else

{

if (payloadBytes.Length > 0)

{

ProcessUpdateQueue(updateMessage);

}

}

}

reader.Close();

}

_monitorQueueCommand.Transaction.Commit();

}

}

catch (Exception ex)

{

if (_monitorQueueCommand.Transaction != null && _monitorQueueCommand.Transaction.Connection != null)

{

// Dequeue the bad message and ends the conversation

SqlCommand cmdQuarantineMessage = new SqlCommand("usp_FailMessage", _ssbConnection, _monitorQueueCommand.Transaction);

cmdQuarantineMessage.CommandType = CommandType.StoredProcedure;

cmdQuarantineMessage.ExecuteNonQuery();

if (reader != null && !reader.IsClosed) reader.Close();

if (_monitorQueueCommand.Transaction != null) _monitorQueueCommand.Transaction.Commit();

}



log.Error("Error monitoring queue.", ex);

}

finally

{

if (reader != null && !reader.IsClosed) reader.Close();

if (_monitorQueueCommand.Transaction != null)

{

_monitorQueueCommand.Transaction = null;

}

}

}

}

}

}


The stored procedure that is monitoring looks like this:


CREATE PROCEDURE [dbo].[asp_ProcessMessage]

AS

BEGIN

WAITFOR ( RECEIVE TOP ( 1 ) conversation_handle, message_type_name, message_body

FROM [MyQueue] ) ;

END