def ReleaseProcessedFlow(self, flow_obj, cursor=None):
    """Releases a flow that the worker was processing to the database."""

    update_query = """
    UPDATE flows
    LEFT OUTER JOIN (
      SELECT client_id, flow_id, needs_processing
      FROM flow_requests
      WHERE
        client_id = %(client_id)s AND
        flow_id = %(flow_id)s AND
        request_id = %(next_request_to_process)s AND
        needs_processing
    ) AS needs_processing
    ON
      flows.client_id = needs_processing.client_id AND
      flows.flow_id = needs_processing.flow_id
    SET
      flows.flow = %(flow)s,
      flows.processing_on = NULL,
      flows.processing_since = NULL,
      flows.processing_deadline = NULL,
      flows.next_request_to_process = %(next_request_to_process)s,
      flows.flow_state = %(flow_state)s,
      flows.user_cpu_time_used_micros = %(user_cpu_time_used_micros)s,
      flows.system_cpu_time_used_micros = %(system_cpu_time_used_micros)s,
      flows.network_bytes_sent = %(network_bytes_sent)s,
      flows.num_replies_sent = %(num_replies_sent)s,
      flows.last_update = NOW(6)
    WHERE
      flows.client_id = %(client_id)s AND
      flows.flow_id = %(flow_id)s AND (
        needs_processing.needs_processing = FALSE OR
        needs_processing.needs_processing IS NULL)
    """

    clone = flow_obj.Copy()
    clone.processing_on = None
    clone.processing_since = None
    clone.processing_deadline = None
    args = {
        "client_id":
            db_utils.ClientIDToInt(flow_obj.client_id),
        "flow":
            clone.SerializeToString(),
        "flow_id":
            db_utils.FlowIDToInt(flow_obj.flow_id),
        "flow_state":
            int(clone.flow_state),
        "network_bytes_sent":
            flow_obj.network_bytes_sent,
        "next_request_to_process":
            flow_obj.next_request_to_process,
        "num_replies_sent":
            flow_obj.num_replies_sent,
        "system_cpu_time_used_micros":
            db_utils.SecondsToMicros(flow_obj.cpu_time_used.system_cpu_time),
        "user_cpu_time_used_micros":
            db_utils.SecondsToMicros(flow_obj.cpu_time_used.user_cpu_time),
    }
    rows_updated = cursor.execute(update_query, args)
    return rows_updated == 1