Piscal Queue working more resilient to error

Core/DAL/DataService.cs

@@ -58,6 +58,17 @@ namespace LeafWeb.Core.DAL
 				select file;
 		}
 
+		public IQueryable<LeafInput> GetRunningLeafInputs()
+		{
+			return
+				from file in _db.LeafInputs
+				where 
+					file.CurrentStatus == LeafInputStatusType.Running ||
+					file.CurrentStatus == LeafInputStatusType.Starting ||
+					file.CurrentStatus == LeafInputStatusType.Finishing
+				select file;
+		}
+
 		public void AddLeafInput(LeafInput leafInput)
 		{
 			leafInput.Added = DateTime.Now;

Core/DAL/LeafWebContext.cs

@@ -2,6 +2,7 @@
 using System.Data.Entity.ModelConfiguration.Conventions;
 using System.Data.Entity.SqlServer;
 using LeafWeb.Core.Entities;
+using NLog;
 
 namespace LeafWeb.Core.DAL
 {
@@ -16,6 +17,8 @@ namespace LeafWeb.Core.DAL
 
 		protected override void OnModelCreating(DbModelBuilder modelBuilder)
 		{
+			LogManager.GetCurrentClassLogger().Debug("OnModelCreating");
+
 			modelBuilder.Conventions.Remove<PluralizingTableNameConvention>();
 			base.OnModelCreating(modelBuilder);
 		}

Core/DAL/LeafWebInitializer.cs

@@ -3,6 +3,7 @@ using System.Linq;
 using LeafWeb.Core.Entities;
 using LeafWeb.Core.Parsers;
 using LeafWeb.Core.Utility;
+using NLog;
 
 namespace LeafWeb.Core.DAL
 {
@@ -12,6 +13,8 @@ namespace LeafWeb.Core.DAL
 
 		protected override void Seed(LeafWebContext context)
 		{
+			LogManager.GetCurrentClassLogger().Debug("Seed");
+
 			// get fluxnet sites from file
 			var fileInfo = FileUtility.GetContentFile(ContentDirectory, "fluxnet_site_list_all_October2015_with_joins_corrected.csv", true);
 			var fluxnetSiteCsvParser = new FluxnetSiteCsvParser(fileInfo);

Core/Entities/LeafInput.cs

@@ -29,7 +29,8 @@ namespace LeafWeb.Core.Entities
 		[Required(ErrorMessage = "Site Id required")]
 		public string SiteId { get; set; }
 
-		[Required(ErrorMessage = "PhotosynthesisType required")]
+		// [Required(ErrorMessage = "PhotosynthesisType required")]
+		// http://stackoverflow.com/questions/6038541/ef-validation-failing-on-update-when-using-lazy-loaded-required-properties
 		public virtual PhotosynthesisType PhotosynthesisType { get; set; }
 
 		[DataType(DataType.Date)]

Core/Entities/LeafInputStatusType.cs

@@ -3,8 +3,10 @@ namespace LeafWeb.Core.Entities
 	public enum LeafInputStatusType
 	{
 		Pending = 0,
-		Running = 1,
+		Starting = 1,
-		Complete = 2,
+		Running = 2,
-		Exception = 3
+		Finishing = 3,
+		Complete = 4,
+		Exception = 5
 	}
 }

Core/Remote/PiscalClientException.cs

@@ -4,8 +4,10 @@ namespace LeafWeb.Core.Remote
 {
 	public class PiscalClientException : Exception
 	{
-		public PiscalClientException(string error) : base(error)
+		public int LeafInputId { get; private set; }
+		public PiscalClientException(int leafInputId, string error) : base(error)
 		{
+			LeafInputId = leafInputId;
 		}
 	}
 }

Core/Remote/PiscalSshClient.cs

@@ -74,7 +74,7 @@ namespace LeafWeb.Core.Remote
 				ssh.Disconnect();
 
 				if (command.ExitStatus != 0)
-					throw new PiscalClientException(command.Result);
+					throw new PiscalClientException(leafInput.LeafInputId, command.Result);
 
 				_logger.Debug("RunLeafInput result: " + command.Result);
 			}
@@ -93,7 +93,7 @@ namespace LeafWeb.Core.Remote
 				case StatusNotStarted:
 					return PiscalStatus.NotStarted;
 				default:
-					throw new PiscalClientException("Unknown status: " + statusRaw[0]);
+					throw new PiscalClientException(leafInput.LeafInputId, "Unknown status: " + statusRaw[0]);
 			}
 		}
 
@@ -108,7 +108,7 @@ namespace LeafWeb.Core.Remote
 				ssh.Disconnect();
 
 				if (command.ExitStatus != 0)
-					throw new PiscalClientException(command.Result);
+					throw new PiscalClientException(leafInput.LeafInputId, command.Result);
 
 				return command.Result
 					.SplitNewLine()
@@ -125,7 +125,7 @@ namespace LeafWeb.Core.Remote
 			// get output files
 			var status = GetLeafInputStatusRaw(leafInput);
 			if (status[0] != StatusComplete)
-				throw new PiscalClientException("output not available, status is " + status[0]);
+				throw new PiscalClientException(leafInput.LeafInputId, "output not available, status is " + status[0]);
 
 			var filePaths = status.Skip(1);
 
@@ -162,7 +162,7 @@ namespace LeafWeb.Core.Remote
 				ssh.Disconnect();
 
 				if (command.ExitStatus != 0)
-					throw new PiscalClientException(command.Error);
+					throw new PiscalClientException(leafInput.LeafInputId, command.Error);
 			}
 		}
 	}

Core/Remote/piscal_launcher.cfg

@@ -0,0 +1,5 @@
+piscal_executable="/home/poprhythm/Code/piscal/leafres/testrun/piscal"
+storage_directory="/home/poprhythm/LeafWebStorage"
+
+# uncomment to use test output instead of running Piscal
+# test_output_directory="/home/poprhythm/LeafWebTestOutput"

Core/Remote/piscal_launcher.sh

@@ -1,28 +1,39 @@
 #!/bin/bash
 # piscal launcher
 
-usage="$(basename "$0") [-h] -d working_directory [-f input_filename] -- script to launch Piscal
+usage="$(basename "$0") [-h] -d working_directory [-u notify_url] -- script to launch Piscal
 
 where:
     -h  show this help text
     -d  working directory
-    -f  input filename"
+    -f  input filename
+    -u  url to notify on completion"
 
-directory=""
+# http://stackoverflow.com/a/246128/99492
-input_filename=""
+base_directory="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+working_directory=""
+output_directory_base_name="output"
+input_directory_name="input"
+cleaned_input_directory_name="clninput"
+touser_directory_name="fitresult/touser"
+nottouser_directory_name="fitresult/nottouser"
+notify_url=""
 
-while getopts "hd:f:" opt; do
+# import the settings from piscal.cfg
+# $piscal_executable and $storage_directory
+. "$base_directory/piscal_launcher.cfg"
+
+while getopts "hd:f:u:" opt; do
     case "$opt" in
     h )
         echo "$usage"
         exit
         ;;
     d )
-        directory=$OPTARG
+        working_directory=$OPTARG
         ;;
-    f )
+    u )
-        input_filename=$OPTARG
+        notify_url=$OPTARG
-        task="process"
         ;;
     \?) printf "illegal option: -%s\n" "$OPTARG" >&2
         echo "$usage" >&2
@@ -30,29 +41,62 @@ while getopts "hd:f:" opt; do
         ;;
     esac
 done
-if [ -z "$directory" ]; then
+
+if [ -z "$working_directory" ]; then
     echo "working directory required (-d)"
     exit 1
 fi
-if [ ! -d "$directory" ]; then
+if [ ! -d "$working_directory" ]; then
-    echo "working directory $directory not found"
+    echo "working directory $working_directory not found"
     exit 1
 fi
-if [ -z "$input_filename" ]; then
+output_directory_base="$working_directory/$output_directory_base_name"
-    echo "input filename required (-f)"
+output_directory_touser="$output_directory_base/$touser_directory_name"
-    exit 1
+output_directory_nottouser="$output_directory_base/$nottouser_directory_name"
+cleaned_input_directory="$output_directory_base/$cleaned_input_directory_name"
+run_directory="$working_directory/run"
+
+# setup output directories
+if [ ! -d "$output_directory_base" ]; then
+    mkdir "$output_directory_base"
 fi
-if [ ! -f "$directory/$input_filename" ]; then
+if [ ! -d "$cleaned_input_directory" ]; then
-    echo "input filename $directory/$input_filename not found"
+    mkdir "$cleaned_input_directory"
-    exit 1
 fi
 
-output_directory="$directory/output"
+#find "$working_directory/$input_directory_name" -maxdepth 1 -type f\
+#    -exec cp {} "$cleaned_input_directory" \;
 
-sleep 1m
+if [ ! -d "$output_directory_base/fitresult" ]; then
-if [ ! -d "$output_directory" ]; then
+    mkdir "$output_directory_base/fitresult"
-    mkdir "$output_directory"
+fi
+if [ ! -d "$output_directory_touser" ]; then
+    mkdir "$output_directory_touser"
+fi
+if [ ! -d "$output_directory_nottouser" ]; then
+    mkdir "$output_directory_nottouser"
+fi
+if [ ! -d "$run_directory" ]; then
+    mkdir "$run_directory"
 fi
 
-# TODO: run actual command
+# run piscal
-cp "/home/poprhythm/LeafWebTestOutput"/* "$output_directory"
+pushd $run_directory >> /dev/null
+
+if [ -z "$test_output_directory" ]; then
+    eval $piscal_executable
+else
+    cp "$test_output_directory"/* "$output_directory_touser"/
+    cp "$test_output_directory"/* "$output_directory_nottouser"/
+fi
+
+popd >> /dev/null
+
+# copy output to storage
+cp "$output_directory_touser"/*.csv "$storage_directory"/
+cp "$output_directory_nottouser"/*.csv "$storage_directory"/
+
+# notify given url of completion
+if [ -z "$notify_url" ]; then
+    wget -qO- "$notify_url" &> /dev/null
+fi

Core/Remote/piscal_manager.sh

@@ -1,27 +1,33 @@
 #!/bin/bash
 # piscal manager script
 
-usage="$(basename "$0") [-h] -d directory_name [-f input_filename|-s] -- script to manage Piscal
+usage="$(basename "$0") [-h] -d directory_name -p photosynthetic_type [-s|-c|-k] -- script to manage Piscal
 
 where:
     -h  show this help text
     -d  working directory name
-    -f  input filename
+    -p  photosynthetic type
-    -s  job status"
+    -s  start job
+    -c  cleanup directory
+    -k  kill job"
+
+# Initialize variables:
+# http://stackoverflow.com/a/246128/99492
+base_directory="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+directory_name=""
+photosynthetic_type="C3_photosynthesis_leafweb" #default
+task="get_status" # default task
+input_directory_name="input"
+pid_filename="piscal.pid"
+stderr_filename="piscal.err"
+output_directory_name="output/fitresult/touser"
+cleaned_input_directory_name="output/clninput"
+launcher="$base_directory/piscal_launcher.sh"
 
 # http://stackoverflow.com/a/14203146/99492
 # http://wiki.bash-hackers.org/howto/getopts_tutorial
 
-# Initialize variables:
+while getopts "hd:f:p:sck" opt; do
-base_directory="/home/poprhythm/LeafWeb"
-directory_name=""
-input_filename=""
-task="start" # default task
-pid_filename="piscal.pid"
-out_filename="piscal.out"
-launcher="$base_directory/piscal_launcher.sh"
-
-while getopts "hd:f:s" opt; do
     #echo "$opt = $OPTARG"
     case "$opt" in
     h )
@@ -31,12 +37,17 @@ while getopts "hd:f:s" opt; do
     d )
         directory_name=$OPTARG
         ;;
-    f )
+    p )
-        input_filename=$OPTARG
+        photosynthetic_type=$OPTARG
-        task="start"
         ;;
     s )
-        task="get_status"
+        task="launch"
+        ;;
+    c )
+        task="cleanup"
+        ;;
+    k )
+        task="kill"
         ;;
     \?) printf "illegal option: -%s\n" "$OPTARG" >&2
         echo "$usage" >&2
@@ -44,39 +55,45 @@ while getopts "hd:f:s" opt; do
         ;;
     esac
 done
+
+## Examine directories
 if [ -z "$directory_name" ]; then
     echo "directory name required (-d)"
     exit 1
 fi
 working_directory="$base_directory/$directory_name"
 if [ ! -d "$working_directory" ]; then
-    echo "directory $working_directory not found"
+    echo "working directory $working_directory not found"
     exit 1
 fi
-if [ "$task" = "start" ]; then
+input_directory="$working_directory/$input_directory_name"
-    if [ -z "$input_filename" ]; then
+if [ ! -d "$input_directory" ]; then
-        echo "input filename required (-f)"
+    echo "input directory $input_directory not found"
     exit 1
-    fi
+fi
+pid_path="$working_directory/$pid_filename"
 
-    if [ ! -f "$working_directory/$input_filename" ]; then
+## Process task
-        echo "input filename $working_directory/$input_filename not found"
+if [ "$task" = "launch" ]; then
-        exit 1
+    piscal_config_file="$working_directory"/piscal.cfg
-    fi
+    # write config file for piscal
+    echo $photosynthetic_type > "$piscal_config_file"
+    find "$input_directory" -maxdepth 1 -type f\
+        -printf '%P\n'\
+        >> "$piscal_config_file"
     
-    command="$launcher -d $working_directory -f $input_filename"
+    command="$launcher -d $working_directory -f piscal.cfg"
-    nohup ${command} > $working_directory/$out_filename 2>&1 &
+    # launch it, sending error output to file
+    nohup ${command} > $working_directory/$stderr_filename 2>&1 &
 
     # write the PID to a temp file to check for completion later
     echo $! > $working_directory/$pid_filename
     echo started
 elif [ "$task" = "get_status" ]; then
-    pid_path="$working_directory/$pid_filename"
+    # if the pid doesn't exist, then process hasn't started
-    
-    # if the pid doesn't exist, then process never started
     if [ ! -f "$pid_path" ]; then
-        echo "no pid file found"
+        echo "not started"
-        exit 1
+        exit
     fi
 
     pid=$(head -n 1 $pid_path)
@@ -85,14 +102,36 @@ elif [ "$task" = "get_status" ]; then
         # if it is in ps, then it's still running
         echo running
     else
-        # otherwise, it is complete, check the output for success/error
+        # otherwise, it is complete, check for runtime errors
-        # TODO: examine output for errors, etc
+        if [ -s "$working_directory/$stderr_filename" ]; then
-        #cat piscal.out
+            cat "$working_directory/$stderr_filename"
-        if [ ! -d "$working_directory/output" ]; then
-            echo "output directory $working_directory/output not found"
             exit 1
         fi
-        echo success
+        
-        find "$working_directory/output"/*
+        output_directory="$working_directory/$output_directory_name"
+        if [ ! -d "$output_directory" ]; then
+            echo "output directory $output_directory not found"
+            exit 1
         fi
+        
+        echo complete
+        find "$output_directory"/*
+    fi
+elif [ "$task" = "cleanup" ]; then
+    rm -rf "$working_directory"
+    echo hi
+elif [ "$task" = "kill" ]; then
+    # if the pid doesn't exist, then process hasn't started
+    if [ ! -f "$pid_path" ]; then
+        echo "not started"
+        exit
+    fi
+
+    pid=$(head -n 1 $pid_path)
+    # if the pid exists, check the process status using ps
+    if ps -p $pid > /dev/null; then
+        # if it is in ps, then it's still running
+        kill $pid
+    fi
+    echo killed
 fi

Web/HangfireStartup.cs

@@ -3,7 +3,6 @@ using Hangfire;
 using Microsoft.Owin;
 using LeafWeb.Web;
 using LeafWeb.Web.Services;
-using NLog.Internal;
 using Owin;
 using ConfigurationManager = System.Configuration.ConfigurationManager;
 
@@ -24,7 +23,6 @@ namespace LeafWeb.Web
 
 		private void SetupRecurringJobs()
 		{
-			// TODO: new SqlServerDistributedLock
 			var queueInterval = ConfigurationManager.AppSettings["ProcessQueueInterval"];
 			// https://discuss.hangfire.io/t/how-to-create-cron-job-that-is-executing-every-15-minutes/533
 			RecurringJob.AddOrUpdate<PiscalQueueManager>(PiscalProcessQueue, p => p.ProcessQueue(), queueInterval);

Web/Services/Cleanup.cs

@@ -0,0 +1,25 @@
+using LeafWeb.Core.Entities;
+using LeafWeb.Core.Remote;
+
+namespace LeafWeb.Web.Services
+{
+	public class Cleanup : PiscalQueueWorker
+	{
+		protected override void DoWorkInternal(LeafInput leafInput)
+		{
+			try
+			{
+				Logger.Info("LeafInput: {0}, Cleanup", leafInput.Id);
+
+				PiscalService.Cleanup(leafInput);
+			}
+			catch (PiscalClientException ex)
+			{
+				var errorMessage = FormatException(ex, ex.LeafInputId);
+				Logger.Error(errorMessage, ex);
+				// log the error, but ignore the cleanup issue for now
+				Logger.Info("LeafInput: {0}, Cleanup - likely has not occurred", leafInput.Id);
+			}
+		}
+	}
+}

Web/Services/PiscalQueueBase.cs

@@ -0,0 +1,57 @@
+using System;
+using Hangfire;
+using LeafWeb.Core.DAL;
+using LeafWeb.Core.Entities;
+using LeafWeb.Core.Remote;
+using NLog;
+
+namespace LeafWeb.Web.Services
+{
+	public abstract class PiscalQueueBase : IDisposable
+	{
+		protected readonly DataService DataService;
+		protected readonly PiscalService PiscalService;
+		protected readonly Logger Logger;
+
+		protected PiscalQueueBase(DataService dataService, PiscalService piscalService)
+		{
+			DataService = dataService;
+			PiscalService = piscalService;
+			Logger = LogManager.GetLogger(GetType().Name);
+		}
+
+		protected PiscalQueueBase() : this(new DataService(), new PiscalService()) { }
+
+		protected string FormatException(Exception ex, int leafInputId)
+		{
+			return
+				$"LeafInput: {leafInputId}{Environment.NewLine}" +
+				$"Class: {GetType().Name}{Environment.NewLine}" + 
+				$"Exception: {ex.Message}{Environment.NewLine}" +
+				(ex.InnerException != null ? $"InnerException: {ex.InnerException}{Environment.NewLine}" : string.Empty)
+				+ $"StackTrace: {ex.StackTrace}";
+		}
+
+		protected void PiscalExceptionNotify(PiscalClientException ex, LeafInput leafInput)
+		{
+			var errorMessage = FormatException(ex, ex.LeafInputId);
+			Logger.Error(errorMessage);
+			BackgroundJob.Enqueue<EmailNotificationService>(
+				email => email.SendAdministratorMessage($"LeafWeb: PiscalQueue {GetType().Name} Exception", errorMessage));
+
+			// TODO send user email too
+
+			if (leafInput != null)
+			{
+				DataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Exception, "Error occurred processing LeafInput",
+					ex.Message);
+				BackgroundJob.Enqueue<Cleanup>(s => s.DoWork(leafInput.Id));
+			}
+		}
+		
+		public void Dispose()
+		{
+			DataService.Dispose();
+		}
+	}
+}

Web/Services/PiscalQueueManager.cs

@@ -2,168 +2,111 @@
 using System.Linq;
 using System.Threading;
 using Hangfire;
-using LeafWeb.Core.DAL;
 using LeafWeb.Core.Entities;
 using LeafWeb.Core.Remote;
-using NLog;
 
 namespace LeafWeb.Web.Services
 {
-	public class PiscalQueueManager : IDisposable
+	public class PiscalQueueManager : PiscalQueueBase
 	{
-		private readonly DataService _dataService;
-		private readonly PiscalService _piscalService;
-		private readonly EmailNotificationService _emailService;
-
-		public PiscalQueueManager(DataService dataService, PiscalService piscalService, EmailNotificationService emailService)
-		{
-			_dataService = dataService;
-			_piscalService = piscalService;
-			_emailService = emailService;
-		}
-
-		public PiscalQueueManager() : this(new DataService(), new PiscalService(), new EmailNotificationService()) {}
-
 		private static readonly object ProcessQueueLock = new object();
 
 		public void ProcessQueue()
 		{
-			var logger = LogManager.GetCurrentClassLogger();
+			// prevent multiple entry into processing the queue
-
 			if (Monitor.TryEnter(ProcessQueueLock))
 			{
-				logger.Trace("ProcessQueue entered");
+				Logger.Trace("ProcessQueue entered");
 
 				try
 				{
-					ProcessRunning(logger);
+					UpdateRunning();
 
-					ProcessQueue(logger);
+					StartNextPending();
 				}
 				finally
 				{
-					logger.Trace("ProcessQueue exit");
+					Logger.Trace("ProcessQueue exit");
 
 					Monitor.Exit(ProcessQueueLock);
 				}
 			}
 			else
 			{
-				logger.Trace("ProcessQueue locked, queue already processing");
+				Logger.Trace("ProcessQueue locked, queue already processing");
 			}
 		}
 	
-		private void ProcessQueue(ILogger logger)
+		private void StartNextPending()
 		{
-			var runningLeafInputs = _dataService.GetLeafInputs(LeafInputStatusType.Running).ToList();
+			var runningLeafInputs = DataService.GetRunningLeafInputs().ToList();
 			if (runningLeafInputs.Any())
 			{
-				logger.Trace("Leaf input currently running , don't enqueue any new");
+				Logger.Trace("Leaf input currently running , don't enqueue any new");
 				return;
 			}
 
-			var pending =
+			var pendingInput = 
-				_dataService
+				DataService
 					.GetLeafInputs(LeafInputStatusType.Pending)
 					.OrderBy(l => l.StatusHistory.Min(sh => sh.DateTime))
 					.FirstOrDefault();
 
-			if (pending == null)
+			if (pendingInput == null)
 			{
-				logger.Trace("No pending leaf input");
+				Logger.Trace("No pending leaf input");
 				return;
 			}
 
-			logger.Info("LeafInputFile: {0}, Start", pending.Id);
+			var pendingInputId = pendingInput.Id;
-			try
+			Logger.Info("LeafInput: {0}, Starting", pendingInputId);
-			{
+			DataService.SetLeafInputStatus(pendingInput, LeafInputStatusType.Starting);
-				_piscalService.Run(pending);
+			BackgroundJob.Enqueue<StartPending>(c => c.DoWork(pendingInputId));
-			}
-			catch (PiscalClientException ex)
-			{
-				var errorMessage = 
-					$"LeafInputFile: {pending.Id}\r\nProcessRunning Exception: {ex.Message}"
-					+ $"\r\nInnerException: {ex.InnerException}\r\nStackTrace: {ex.StackTrace}";
-				logger.Error(errorMessage);
-				BackgroundJob.Enqueue(() => _emailService.SendAdministratorMessage("LeafWeb ProcessQueue Exception", errorMessage));
-
-				_dataService.SetLeafInputStatus(pending, LeafInputStatusType.Exception, "Error occurred starting LeafInput", ex.Message);
-				logger.Info("LeafInputFile: {0}, Cleanup", pending.Id);
-				_piscalService.Cleanup(pending);
-
-				// TODO: re-queue?
-				//_dataService.SetLeafInputFileStatus(queuedFile, LeafInputStatusType.Queued, "Re-queuing LeafInput");
-			}
-			logger.Trace("LeafInputFile: {0}, Set Pending", pending.Id);
-			_dataService.SetLeafInputStatus(pending, LeafInputStatusType.Running);
 		}
 
-		private void ProcessRunning(ILogger logger)
+		private void UpdateRunning()
 		{
-			var running = _dataService.GetLeafInputs(LeafInputStatusType.Running).ToList();
+			var running = DataService.GetLeafInputs(LeafInputStatusType.Running).ToList();
 			foreach (var leafInput in running)
 			{
 				try
 				{
-					var status = _piscalService.GetStatus(leafInput);
+					var status = PiscalService.GetStatus(leafInput);
+					var leafInputId = leafInput.Id;
 					switch (status)
 					{
 						case PiscalStatus.NotStarted:
-							logger.Warn("LeafInputFile: {0}, Not Started, re-queueing", leafInput.Id);
+							// if it's not started - this is unusual state
-							// if it's not started, try to requeue the process - this is unusual state
+							Logger.Warn("LeafInput: {0}, Piscal Not Started", leafInput.Id);
-							_dataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Pending);
 							break;
 
 						case PiscalStatus.Running:
-							logger.Trace("LeafInputFile: {0}, Running", leafInput.Id);
+							Logger.Trace("LeafInput: {0}, Piscal Running", leafInput.Id);
 							// continue running
 							break;
 
 						case PiscalStatus.Complete:
-							logger.Info("LeafInputFile: {0}, Complete", leafInput.Id);
+							// TODO: change to "retrieving output"?
-							// collect the leaf output
+							DataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Finishing);
-							var leafOutputFiles = _piscalService.RetrieveOutputFiles(leafInput).ToList();
 
-							logger.Trace("LeafInputFile: {0}, saving output files", leafInput.Id);
+							var retrieveFilesId = BackgroundJob.Enqueue<RetrieveOutputFiles>(s => s.DoWork(leafInputId));
-							foreach (var outputFile in leafOutputFiles)
+							BackgroundJob.ContinueWith<Cleanup>(retrieveFilesId, s => s.DoWork(leafInputId));
-								_dataService.AddLeafOutputFile(outputFile);
+							BackgroundJob.ContinueWith<EmailNotificationService>(retrieveFilesId,
-
+								email => email.SendLeafWebComplete(leafInputId));
-							logger.Info("LeafInputFile: {0}, output files: {1}", leafInput.Id,
+							BackgroundJob.ContinueWith<SetComplete>(retrieveFilesId, s => s.DoWork(leafInputId));
-								string.Join(", ", leafOutputFiles.Select(o => o.Filename)));
-
-							// update db
-							logger.Trace("LeafInputFile: {0}, Set Complete", leafInput.Id);
-							_dataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Complete);
-
-							BackgroundJob.Enqueue(() => _emailService.SendLeafWebComplete(leafInput.Id));
-
-							// remove working data from the server
-							logger.Info("LeafInputFile: {0}, Cleanup", leafInput.Id);
-							_piscalService.Cleanup(leafInput);
 							break;
 					}
 				}
+				catch (PiscalClientException ex)
+				{
+					PiscalExceptionNotify(ex, leafInput);
+				}
 				catch (Exception ex)
 				{
-					var errorMessage = 
+					var errorMessage = FormatException(ex, leafInput.Id);
-						$"LeafInputFile: {leafInput.Id}\r\nProcessRunning Exception: {ex.Message}" 
+					Logger.Error(errorMessage);
-						+ $"\r\nInnerException: {ex.InnerException}\r\nStackTrace: {ex.StackTrace}";
-					logger.Error(errorMessage);
-					BackgroundJob.Enqueue(() => _emailService.SendAdministratorMessage("LeafWeb ProcessRunning Exception", errorMessage));
-
-					_dataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Exception, "Error occurred processing LeafInput", ex.Message);
-
-					// TODO: Send email
-					logger.Info("LeafInputFile: {0}, Cleanup", leafInput.Id);
-					_piscalService.Cleanup(leafInput);
-					// TODO: re-queue ?
 				}
 			}
 		}
-
-		public void Dispose()
-		{
-			_dataService.Dispose();
-		}
 	}
 }

Web/Services/PiscalQueueWorker.cs

@@ -0,0 +1,39 @@
+using System;
+using LeafWeb.Core.Entities;
+using LeafWeb.Core.Remote;
+
+namespace LeafWeb.Web.Services
+{
+	public abstract class PiscalQueueWorker : PiscalQueueBase
+	{
+		public void DoWork(int leafInputId)
+		{
+			LeafInput leafInput = null;
+			try
+			{
+				leafInput = DataService.GetLeafInput(leafInputId);
+				DoWorkInternal(leafInput);
+			}
+			catch (PiscalClientException ex)
+			{
+				PiscalExceptionNotify(ex, leafInput);
+
+				if (leafInput != null)
+				{
+					DataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Exception, "Error occurred processing LeafInput",
+						ex.Message);
+				}
+				// signal to process next item
+				HangfireStartup.TriggerPiscalProcessQueue();
+			}
+			catch (Exception ex)
+			{
+				var errorMessage = FormatException(ex, leafInputId);
+				Logger.Error(errorMessage);
+				throw; // this will retry via HangFire
+			}
+		}
+
+		protected abstract void DoWorkInternal(LeafInput leafInputId);
+	}
+}

Web/Services/PiscalService.cs

@@ -5,6 +5,9 @@ using LeafWeb.Core.Remote;
 
 namespace LeafWeb.Web.Services
 {
+	/// <summary>
+	/// Thin layer over PiscalClient to translate Core entities to Piscal objects
+	/// </summary>
 	public class PiscalService
 	{
 		private readonly IPiscalClient _piscalClient;

Web/Services/RetrieveOutputFiles.cs

@@ -0,0 +1,28 @@
+using System.Linq;
+using LeafWeb.Core.Entities;
+
+namespace LeafWeb.Web.Services
+{
+	public class RetrieveOutputFiles : PiscalQueueWorker
+	{
+		protected override void DoWorkInternal(LeafInput leafInput)
+		{
+			Logger.Trace("LeafInput: {0}, RetrieveOutputFiles", leafInput.Id);
+
+			var leafOutputFiles = PiscalService.RetrieveOutputFiles(leafInput).ToList();
+
+			Logger.Trace("LeafInput: {0}, RetrieveOutputFiles saving output files", leafInput.Id);
+
+			foreach (var outputFile in leafOutputFiles)
+			{
+				if (leafInput.OutputFiles.All(file => file.Filename != outputFile.Filename))
+					DataService.AddLeafOutputFile(outputFile);
+				else
+					Logger.Warn("LeafInput: {0}, RetrieveOutputFiles duplicate file name: {1}", leafInput.Id, outputFile.Filename);
+			}
+
+			Logger.Info("LeafInput: {0}, RetrieveOutputFiles output files: {1}", leafInput.Id,
+				string.Join(", ", leafOutputFiles.Select(o => o.Filename)));
+		}
+	}
+}

Web/Services/SetComplete.cs

@@ -0,0 +1,16 @@
+using LeafWeb.Core.Entities;
+
+namespace LeafWeb.Web.Services
+{
+	public class SetComplete : PiscalQueueWorker
+	{
+		protected override void DoWorkInternal(LeafInput leafInput)
+		{
+			Logger.Trace("LeafInput: {0}, Set Complete", leafInput.Id);
+
+			DataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Complete);
+
+			HangfireStartup.TriggerPiscalProcessQueue();
+		}
+	}
+}

Web/Services/StartPending.cs

@@ -0,0 +1,16 @@
+using LeafWeb.Core.Entities;
+
+namespace LeafWeb.Web.Services
+{
+	public class StartPending : PiscalQueueWorker
+	{
+		protected override void DoWorkInternal(LeafInput leafInput)
+		{
+			Logger.Trace("LeafInput: {0}, Run", leafInput.Id);
+
+			PiscalService.Run(leafInput);
+
+			DataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Running);
+		}
+	}
+}

Web/Web.csproj

@@ -960,10 +960,16 @@
     </Compile>
     <Compile Include="Attributes\HttpParamActionAttribute.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="Services\Cleanup.cs" />
     <Compile Include="Services\EmailNotificationService.cs" />
+    <Compile Include="Services\PiscalQueueBase.cs" />
     <Compile Include="Services\PiscalQueueManager.cs" />
+    <Compile Include="Services\PiscalQueueWorker.cs" />
     <Compile Include="Services\PiscalService.cs" />
     <Compile Include="HangfireStartup.cs" />
+    <Compile Include="Services\RetrieveOutputFiles.cs" />
+    <Compile Include="Services\SetComplete.cs" />
+    <Compile Include="Services\StartPending.cs" />
     <Compile Include="Utility\MarkdownHelper.cs" />
     <Compile Include="Utility\Validation.cs" />
     <Compile Include="ViewModels\LeafInput\ConfirmViewModel.cs" />