[#1032] Error out torrent if data is missing on startup

4 files changed, 256 insertions, 66 deletions

diff --git a/deluge/core/torrent.py b/deluge/core/torrent.py
index c6031acf8..4583c6611 100644
--- a/deluge/core/torrent.py
+++ b/deluge/core/torrent.py
@@ -169,6 +169,13 @@ class TorrentOptions(dict):
         self["seed_mode"] = False
 
 
+class TorrentError(object):
+    def __init__(self, error_message, was_paused=False, restart_to_resume=False):
+        self.error_message = error_message
+        self.was_paused = was_paused
+        self.restart_to_resume = restart_to_resume
+
+
 class Torrent(object):
     """Torrent holds information about torrents added to the libtorrent session.
 
@@ -196,7 +203,7 @@ class Torrent(object):
         options (dict): The torrent options.
         filename (str): The filename of the torrent file in case it is required.
         is_finished (bool): Keep track if torrent is finished to prevent some weird things on state load.
-        statusmsg (str): Status message holds error info about the torrent
+        statusmsg (str): Status message holds error/extra info about the torrent.
         state (str): The torrent's state
         trackers (list of dict): The torrent's trackers
         tracker_status (str): Status message of currently connected tracker
@@ -204,6 +211,7 @@ class Torrent(object):
         forcing_recheck (bool): Keep track if we're forcing a recheck of the torrent
         forcing_recheck_paused (bool): Keep track if we're forcing a recheck of the torrent so that
             we can re-pause it after its done if necessary
+        forced_error (TorrentError): Keep track if we have forced this torrent to be in Error state.
     """
     def __init__(self, handle, options, state=None, filename=None, magnet=None):
         self.torrent_id = str(handle.info_hash())
@@ -235,18 +243,13 @@ class Torrent(object):
             self.set_trackers(state.trackers)
             self.is_finished = state.is_finished
             self.filename = state.filename
-            last_sess_prepend = "[Error from Previous Session] "
-            if state.error_statusmsg and not state.error_statusmsg.startswith(last_sess_prepend):
-                self.error_statusmsg = last_sess_prepend + state.error_statusmsg
-            else:
-                self.error_statusmsg = state.error_statusmsg
         else:
             self.set_trackers()
             self.is_finished = False
             self.filename = filename
-            self.error_statusmsg = None
 
-        self.statusmsg = "OK"
+        self.forced_error = None
+        self.statusmsg = None
         self.state = None
         self.moving_storage = False
         self.moving_storage_dest_path = None
@@ -610,18 +613,15 @@ class Torrent(object):
         status = self.handle.status()
         session_paused = component.get("Core").session.is_paused()
         old_state = self.state
-        if status.error or self.error_statusmsg:
+
+        if self.forced_error:
+            self.state = "Error"
+            self.set_status_message(self.forced_error.error_message)
+        elif status.error:
             self.state = "Error"
-            # This will be reverted upon resuming.
+            # auto-manage status will be reverted upon resuming.
             self.handle.auto_managed(False)
-            if not status.paused:
-                self.handle.pause()
-            if status.error:
-                self.set_error_statusmsg(decode_string(status.error))
-                log.debug("Error state from lt: %s", self.error_statusmsg)
-            else:
-                log.debug("Error state forced by Deluge, error_statusmsg: %s", self.error_statusmsg)
-            self.set_status_message(self.error_statusmsg)
+            self.set_status_message(decode_string(status.error))
         elif self.moving_storage:
             self.state = "Moving"
         elif not session_paused and status.paused and status.auto_managed:
@@ -636,29 +636,52 @@ class Torrent(object):
 
         if log.isEnabledFor(logging.DEBUG):
             log.debug("State from lt was: %s | Session is paused: %s\nTorrent state set from '%s' to '%s' (%s)",
-                      status.state, session_paused, old_state, self.state, self.torrent_id)
+                      "error" if status.error else status.state, session_paused, old_state, self.state, self.torrent_id)
+            if self.forced_error:
+                log.debug("Torrent Error state message: %s", self.forced_error.error_message)
 
-    def set_status_message(self, message):
+    def set_status_message(self, message=None):
         """Sets the torrent status message.
 
+        Calling method without a message will reset the message to 'OK'.
+
         Args:
-            message (str): The status message.
+            message (str, optional): The status message.
 
         """
+        if not message:
+            message = "OK"
         self.statusmsg = message
 
-    def set_error_statusmsg(self, message):
-        """Sets the torrent error status message.
+    def force_error_state(self, message, restart_to_resume=True):
+        """Forces the torrent into an error state.
 
-        Note:
-            This will force a torrent into an error state. It is used for
-            setting those errors that are not covered by libtorrent.
+        For setting an error state not covered by libtorrent.
 
         Args:
             message (str): The error status message.
-
+            restart_to_resume (bool, optional): Prevent resuming clearing the error, only restarting
+                session can resume.
         """
-        self.error_statusmsg = message
+        status = self.handle.status()
+        self.handle.auto_managed(False)
+        self.forced_error = TorrentError(message, status.paused, restart_to_resume)
+        if not status.paused:
+            self.handle.pause()
+        self.update_state()
+
+    def clear_forced_error_state(self):
+        if not self.forced_error:
+            return
+
+        if self.forced_error.restart_to_resume:
+            log.error("Restart deluge to clear this torrent error")
+
+        if not self.force_error.was_paused and self.options["auto_managed"]:
+            self.handle.auto_managed(True)
+        self.force_error = None
+        self.set_status_message()
+        self.update_state()
 
     def get_eta(self):
         """Get the ETA for this torrent.
@@ -1018,7 +1041,9 @@ class Torrent(object):
         """
         # Turn off auto-management so the torrent will not be unpaused by lt queueing
         self.handle.auto_managed(False)
-        if self.status.paused:
+        if self.state == "Error":
+            return False
+        elif self.status.paused:
             # This torrent was probably paused due to being auto managed by lt
             # Since we turned auto_managed off, we should update the state which should
             # show it as 'Paused'.  We need to emit a torrent_paused signal because
@@ -1036,28 +1061,26 @@ class Torrent(object):
     def resume(self):
         """Resumes this torrent."""
         if self.status.paused and self.status.auto_managed:
-            log.debug("Torrent is being auto-managed, cannot resume!")
-            return
-
-        # Reset the status message just in case of resuming an Error'd torrent
-        self.set_status_message("OK")
-        self.set_error_statusmsg(None)
-
-        if self.status.is_finished:
-            # If the torrent has already reached it's 'stop_seed_ratio' then do not do anything
-            if self.options["stop_at_ratio"]:
-                if self.get_ratio() >= self.options["stop_ratio"]:
-                    # XXX: This should just be returned in the RPC Response, no event
-                    return
+            log.debug("Resume not possible for auto-managed torrent!")
+        elif self.forced_error and self.forced_error.was_paused:
+            log.debug("Resume skipped for error'd torrent as it was originally paused.")
+        elif (self.status.is_finished and self.options["stop_at_ratio"] and
+                self.get_ratio() >= self.options["stop_ratio"]):
+            log.debug("Resume skipped for torrent as it has reached 'stop_seed_ratio'.")
+        else:
+            # Check if torrent was originally being auto-managed.
+            if self.options["auto_managed"]:
+                self.handle.auto_managed(True)
+            try:
+                self.handle.resume()
+            except RuntimeError as ex:
+                log.debug("Unable to resume torrent: %s", ex)
 
-        if self.options["auto_managed"]:
-            # This torrent is to be auto-managed by lt queueing
-            self.handle.auto_managed(True)
-
-        try:
-            self.handle.resume()
-        except RuntimeError as ex:
-            log.debug("Unable to resume torrent: %s", ex)
+        # Clear torrent error state.
+        if self.forced_error and not self.forced_error.restart_to_resume:
+            self.clear_forced_error_state()
+        elif self.state == "Error" and not self.forced_error:
+            self.handle.clear_error()
 
     def connect_peer(self, peer_ip, peer_port):
         """Manually add a peer to the torrent
@@ -1123,8 +1146,15 @@ class Torrent(object):
             None: The response with resume data is returned in a libtorrent save_resume_data_alert.
 
         """
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug("Requesting save_resume_data for torrent: %s", self.torrent_id)
         flags = lt.save_resume_flags_t.flush_disk_cache if flush_disk_cache else 0
-        self.handle.save_resume_data(flags)
+        # Don't generate fastresume data if torrent is in a Deluge Error state.
+        if self.forced_error:
+            component.get("TorrentManager").waiting_on_resume_data[self.torrent_id].errback(
+                UserWarning("Skipped creating resume_data while in Error state"))
+        else:
+            self.handle.save_resume_data(flags)
 
     def write_torrentfile(self, filedump=None):
         """Writes the torrent file to the state dir and optional 'copy of' dir.
@@ -1135,6 +1165,7 @@ class Torrent(object):
         """
 
         def write_file(filepath, filedump):
+            """Write out the torrent file"""
             log.debug("Writing torrent file to: %s", filepath)
             try:
                 with open(filepath, "wb") as save_file:
@@ -1195,16 +1226,20 @@ class Torrent(object):
 
     def force_recheck(self):
         """Forces a recheck of the torrent's pieces"""
-        paused = self.status.paused
+        if self.forced_error:
+            self.forcing_recheck_paused = self.forced_error.was_paused
+            self.clear_forced_error_state(update_state=False)
+        else:
+            self.forcing_recheck_paused = self.status.paused
+
         try:
             self.handle.force_recheck()
             self.handle.resume()
+            self.forcing_recheck = True
         except RuntimeError as ex:
             log.debug("Unable to force recheck: %s", ex)
-            return False
-        self.forcing_recheck = True
-        self.forcing_recheck_paused = paused
-        return True
+            self.forcing_recheck = False
+        return self.forcing_recheck
 
     def rename_files(self, filenames):
         """Renames files in the torrent.
diff --git a/deluge/core/torrentmanager.py b/deluge/core/torrentmanager.py
index dc9375b0d..fe2f5e503 100644
--- a/deluge/core/torrentmanager.py
+++ b/deluge/core/torrentmanager.py
@@ -52,7 +52,6 @@ class TorrentState:
                  queue=None,
                  auto_managed=True,
                  is_finished=False,
-                 error_statusmsg=None,
                  stop_ratio=2.00,
                  stop_at_ratio=False,
                  remove_at_ratio=False,
@@ -406,6 +405,10 @@ class TorrentManager(component.Component):
 
         component.resume("AlertManager")
 
+        # Store the orignal resume_data, in case of errors.
+        if resume_data:
+            self.resume_data[torrent.torrent_id] = resume_data
+
         # Add to queued torrents set.
         self.queued_torrents.add(torrent.torrent_id)
         if self.config["queue_new_to_top"]:
@@ -604,7 +607,9 @@ class TorrentManager(component.Component):
         for torrent in self.torrents.values():
             if self.session.is_paused():
                 paused = torrent.handle.is_paused()
-            elif torrent.state in ["Paused", "Error"]:
+            elif torrent.forced_error:
+                paused = torrent.forced_error.was_paused
+            elif torrent.state == "Paused":
                 paused = True
             else:
                 paused = False
@@ -626,7 +631,6 @@ class TorrentManager(component.Component):
                 torrent.get_queue_position(),
                 torrent.options["auto_managed"],
                 torrent.is_finished,
-                torrent.error_statusmsg,
                 torrent.options["stop_ratio"],
                 torrent.options["stop_at_ratio"],
                 torrent.options["remove_at_ratio"],
@@ -1022,10 +1026,8 @@ class TorrentManager(component.Component):
             return
         # Set an Error message and pause the torrent
         alert_msg = decode_string(alert.message()).split(':', 1)[1].strip()
-        torrent.set_error_statusmsg("Failed to move download folder: %s" % alert_msg)
         torrent.moving_storage = False
-        torrent.pause()
-        torrent.update_state()
+        torrent.force_error_state("Failed to move download folder: %s" % alert_msg)
 
         if torrent_id in self.waiting_on_finish_moving:
             self.waiting_on_finish_moving.remove(torrent_id)
@@ -1069,7 +1071,6 @@ class TorrentManager(component.Component):
             torrent_id = str(alert.handle.info_hash())
         except RuntimeError:
             return
-
         if torrent_id in self.torrents:
             # libtorrent add_torrent expects bencoded resume_data.
             self.resume_data[torrent_id] = lt.bencode(alert.resume_data)
@@ -1090,7 +1091,8 @@ class TorrentManager(component.Component):
 
     def on_alert_fastresume_rejected(self, alert):
         """Alert handler for libtorrent fastresume_rejected_alert"""
-        log.warning("on_alert_fastresume_rejected: %s", decode_string(alert.message()))
+        alert_msg = decode_string(alert.message())
+        log.error("on_alert_fastresume_rejected: %s", alert_msg)
         try:
             torrent_id = str(alert.handle.info_hash())
             torrent = self.torrents[torrent_id]
@@ -1103,8 +1105,8 @@ class TorrentManager(component.Component):
             else:
                 error_msg = "Missing or invalid torrent data!"
         else:
-            error_msg = "Problem with resume data: %s" % decode_string(alert.message()).split(':', 1)[1].strip()
-        torrent.force_error_state(error_msg)
+            error_msg = "Problem with resume data: %s" % alert_msg.split(":", 1)[1].strip()
+        torrent.force_error_state(error_msg, restart_to_resume=True)
 
     def on_alert_file_renamed(self, alert):
         """Alert handler for libtorrent file_renamed_alert
diff --git a/deluge/tests/test_torrent.file.torrent b/deluge/tests/test_torrent.file.torrent
new file mode 100644
index 000000000..fca14ca62
--- /dev/null
+++ b/deluge/tests/test_torrent.file.torrent
@@ -0,0 +1,2 @@
+d7:comment17:Test torrent file10:created by11:Deluge Team13:creation datei1411826665e8:encoding5:UTF-84:infod6:lengthi512000e4:name17:test_torrent.file12:piece lengthi32768e6:pieces320:��$2��Tj
�����>h�U.���--~�–ޔBzu����EB��1����@�ͥ�/K���"z�F�0�֣�[asV�1�B^Wp-�S��׶��F`�M9�)}��,���4n�iW
���j��Q�I����̗(�,�t؋c�h�i��*M��}^�WS���7��	�h:-��b��e�TXa3��m|����J�"���]�0$��}��������l��@L� ��V��,�4�˫���zML�ģJ�*�
+\�AP�&I�9��}�20�֎H��:�_���8�<V�2JYb2�'�2h��0\��*_�j�7:privatei0eee
\ No newline at end of file
diff --git a/deluge/tests/test_torrent_error.py b/deluge/tests/test_torrent_error.py
new file mode 100644
index 000000000..81d0c40e8
--- /dev/null
+++ b/deluge/tests/test_torrent_error.py
@@ -0,0 +1,151 @@
+from __future__ import print_function
+
+import base64
+import os
+import sys
+import time
+
+from twisted.internet import reactor
+from twisted.internet.task import deferLater
+from twisted.trial import unittest
+
+import deluge.component as component
+import deluge.core.torrent
+import deluge.tests.common as common
+from deluge._libtorrent import lt
+from deluge.core.core import Core
+from deluge.core.rpcserver import RPCServer
+from deluge.core.torrentmanager import TorrentState
+
+config_setup = False
+core = None
+rpcserver = None
+eventmanager = None
+
+
+# This is called by torrent.py when calling component.get("...")
+def get(key):
+    if key is "Core":
+        return core
+    elif key is "RPCServer":
+        return rpcserver
+    elif key is "EventManager":
+        return core.eventmanager
+    elif key is "TorrentManager":
+        return core.torrentmanager
+    else:
+        return None
+
+
+class TorrentTestCase(unittest.TestCase):
+
+    def setup_config(self):
+        global config_setup
+        config_setup = True
+        config_dir = common.set_tmp_config_dir()
+        core_config = deluge.config.Config("core.conf", defaults=deluge.core.preferencesmanager.DEFAULT_PREFS,
+                                           config_dir=config_dir)
+        core_config.save()
+
+    def setUp(self):  # NOQA
+        # Save component and set back on teardown
+        self.original_component = deluge.core.torrent.component
+        deluge.core.torrent.component = sys.modules[__name__]
+        self.setup_config()
+        global rpcserver
+        global core
+        rpcserver = RPCServer(listen=False)
+        core = Core()
+        return component.start()
+
+    def tearDown(self):  # NOQA
+        deluge.core.torrent.component = self.original_component
+
+        def on_shutdown(result):
+            component._ComponentRegistry.components = {}
+        return component.shutdown().addCallback(on_shutdown)
+
+    def assert_state(self, torrent, state):
+        torrent.update_state()
+        self.assertEquals(torrent.state, state)
+
+    def get_torrent_atp(self, filename):
+        filename = os.path.join(os.path.dirname(__file__), filename)
+        e = lt.bdecode(open(filename, 'rb').read())
+        info = lt.torrent_info(e)
+        atp = {"ti": info}
+        atp["save_path"] = os.getcwd()
+        atp["storage_mode"] = lt.storage_mode_t.storage_mode_sparse
+        atp["add_paused"] = False
+        atp["auto_managed"] = True
+        atp["duplicate_is_error"] = True
+        return atp
+
+    def test_torrent_error_data_missing(self):
+        options = {"seed_mode": True}
+        filename = os.path.join(os.path.dirname(__file__), "test_torrent.file.torrent")
+        torrent_id = core.add_torrent_file(filename, base64.encodestring(open(filename).read()), options)
+        torrent = core.torrentmanager.torrents[torrent_id]
+
+        self.assert_state(torrent, "Seeding")
+
+        # Force an error by reading (non-existant) piece from disk
+        torrent.handle.read_piece(0)
+        time.sleep(0.2)  # Delay to wait for alert from lt
+        self.assert_state(torrent, "Error")
+
+    def test_torrent_error_resume_original_state(self):
+        options = {"seed_mode": True, "add_paused": True}
+        filename = os.path.join(os.path.dirname(__file__), "test_torrent.file.torrent")
+        torrent_id = core.add_torrent_file(filename, base64.encodestring(open(filename).read()), options)
+        torrent = core.torrentmanager.torrents[torrent_id]
+
+        orig_state = "Paused"
+        self.assert_state(torrent, orig_state)
+
+        # Force an error by reading (non-existant) piece from disk
+        torrent.handle.read_piece(0)
+        time.sleep(0.2)  # Delay to wait for alert from lt
+        self.assert_state(torrent, "Error")
+
+        # Clear error and verify returned to original state
+        torrent.force_recheck()
+        return deferLater(reactor, 0.1, self.assert_state, torrent, orig_state)
+
+    def test_torrent_error_resume_data_unaltered(self):
+        resume_data = {'active_time': 13399L, 'num_incomplete': 16777215L, 'announce_to_lsd': 1L, 'seed_mode': 0L,
+                       'pieces': '\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01', 'paused': 0L,
+                       'seeding_time': 13399L, 'last_scrape': 13399L,
+                       'info-hash': '-\xc5\xd0\xe7\x1af\xfeid\x9ad\r9\xcb\x00\xa2YpIs', 'max_uploads': 16777215L,
+                       'max_connections': 16777215L, 'num_downloaders': 16777215L, 'total_downloaded': 0L,
+                       'file-format': 'libtorrent resume file', 'peers6': '', 'added_time': 1411826665L,
+                       'banned_peers6': '', 'file_priority': [1L], 'last_seen_complete': 0L, 'total_uploaded': 0L,
+                       'piece_priority': '\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01\x01',
+                       'file-version': 1L, 'announce_to_dht': 1L, 'auto_managed': 1L, 'upload_rate_limit': 0L,
+                       'completed_time': 1411826665L, 'allocation': 'sparse', 'blocks per piece': 2L,
+                       'download_rate_limit': 0L, 'libtorrent-version': '0.16.17.0', 'banned_peers': '',
+                       'num_seeds': 16777215L, 'sequential_download': 0L, 'announce_to_trackers': 1L,
+                       'peers': '\n\x00\x02\x0f=\xc6SC\x17]\xd8}\x7f\x00\x00\x01=\xc6', 'finished_time': 13399L,
+                       'last_upload': 13399L, 'trackers': [[]], 'super_seeding': 0L,
+                       'file sizes': [[512000L, 1411826586L]], 'last_download': 13399L}
+        torrent_state = TorrentState(
+            torrent_id='2dc5d0e71a66fe69649a640d39cb00a259704973',
+            filename='test_torrent.file.torrent',
+            name='',
+            save_path='/home/ubuntu/Downloads',
+            file_priorities=[1],
+            is_finished=True,
+        )
+
+        filename = os.path.join(os.path.dirname(__file__), "test_torrent.file.torrent")
+        filedump = open(filename).read()
+        torrent_id = core.torrentmanager.add(state=torrent_state, filedump=filedump,
+                                             resume_data=lt.bencode(resume_data))
+        torrent = core.torrentmanager.torrents[torrent_id]
+
+        def assert_resume_data():
+            self.assert_state(torrent, "Error")
+            tm_resume_data = lt.bdecode(core.torrentmanager.resume_data[torrent.torrent_id])
+            self.assertEquals(tm_resume_data, resume_data)
+
+        return deferLater(reactor, 0.5, assert_resume_data)