diff --git a/homeassistant/components/duckdns/__init__.py b/homeassistant/components/duckdns/__init__.py
index 7d677580177..171d17faff9 100644
--- a/homeassistant/components/duckdns/__init__.py
+++ b/homeassistant/components/duckdns/__init__.py
@@ -1,13 +1,17 @@
 """Integrate with DuckDNS."""
-from datetime import timedelta
 import logging
+from asyncio import iscoroutinefunction
+from datetime import timedelta
 
 import voluptuous as vol
 
-from homeassistant.const import CONF_ACCESS_TOKEN, CONF_DOMAIN
 import homeassistant.helpers.config_validation as cv
-from homeassistant.helpers.event import async_track_time_interval
+from homeassistant.const import CONF_ACCESS_TOKEN, CONF_DOMAIN
+from homeassistant.core import callback, CALLBACK_TYPE
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.event import async_track_point_in_utc_time
+from homeassistant.loader import bind_hass
+from homeassistant.util import dt as dt_util
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -42,25 +46,28 @@ async def async_setup(hass, config):
     token = config[DOMAIN][CONF_ACCESS_TOKEN]
     session = async_get_clientsession(hass)
 
-    result = await _update_duckdns(session, domain, token)
-
-    if not result:
-        return False
-
-    async def update_domain_interval(now):
+    async def update_domain_interval(_now):
         """Update the DuckDNS entry."""
-        await _update_duckdns(session, domain, token)
+        return await _update_duckdns(session, domain, token)
+
+    intervals = (
+        INTERVAL,
+        timedelta(minutes=1),
+        timedelta(minutes=5),
+        timedelta(minutes=15),
+        timedelta(minutes=30),
+    )
+    async_track_time_interval_backoff(hass, update_domain_interval, intervals)
 
     async def update_domain_service(call):
         """Update the DuckDNS entry."""
         await _update_duckdns(session, domain, token, txt=call.data[ATTR_TXT])
 
-    async_track_time_interval(hass, update_domain_interval, INTERVAL)
     hass.services.async_register(
         DOMAIN, SERVICE_SET_TXT, update_domain_service, schema=SERVICE_TXT_SCHEMA
     )
 
-    return result
+    return True
 
 
 _SENTINEL = object()
@@ -89,3 +96,37 @@ async def _update_duckdns(session, domain, token, *, txt=_SENTINEL, clear=False)
         return False
 
     return True
+
+
+@callback
+@bind_hass
+def async_track_time_interval_backoff(hass, action, intervals) -> CALLBACK_TYPE:
+    """Add a listener that fires repetitively at every timedelta interval."""
+    if not iscoroutinefunction:
+        _LOGGER.error("action needs to be a coroutine and return True/False")
+        return
+
+    if not isinstance(intervals, (list, tuple)):
+        intervals = (intervals,)
+    remove = None
+    failed = 0
+
+    async def interval_listener(now):
+        """Handle elapsed intervals with backoff."""
+        nonlocal failed, remove
+        try:
+            failed += 1
+            if await action(now):
+                failed = 0
+        finally:
+            delay = intervals[failed] if failed < len(intervals) else intervals[-1]
+            remove = async_track_point_in_utc_time(hass, interval_listener, now + delay)
+
+    hass.async_run_job(interval_listener, dt_util.utcnow())
+
+    def remove_listener():
+        """Remove interval listener."""
+        if remove:
+            remove()  # pylint: disable=not-callable
+
+    return remove_listener
diff --git a/tests/components/duckdns/test_init.py b/tests/components/duckdns/test_init.py
index 0fdfebac66e..0213d9aefa6 100644
--- a/tests/components/duckdns/test_init.py
+++ b/tests/components/duckdns/test_init.py
@@ -1,28 +1,29 @@
 """Test the DuckDNS component."""
-import asyncio
 from datetime import timedelta
-
+import logging
 import pytest
 
 from homeassistant.loader import bind_hass
 from homeassistant.setup import async_setup_component
 from homeassistant.components import duckdns
 from homeassistant.util.dt import utcnow
+from homeassistant.components.duckdns import async_track_time_interval_backoff
 
 from tests.common import async_fire_time_changed
 
 DOMAIN = "bla"
 TOKEN = "abcdefgh"
+_LOGGER = logging.getLogger(__name__)
+INTERVAL = duckdns.INTERVAL
 
 
 @bind_hass
-@asyncio.coroutine
-def async_set_txt(hass, txt):
+async def async_set_txt(hass, txt):
     """Set the txt record. Pass in None to remove it.
 
     This is a legacy helper method. Do not use it for new tests.
     """
-    yield from hass.services.async_call(
+    await hass.services.async_call(
         duckdns.DOMAIN, duckdns.SERVICE_SET_TXT, {duckdns.ATTR_TXT: txt}, blocking=True
     )
 
@@ -41,40 +42,60 @@ def setup_duckdns(hass, aioclient_mock):
     )
 
 
-@asyncio.coroutine
-def test_setup(hass, aioclient_mock):
+async def test_setup(hass, aioclient_mock):
     """Test setup works if update passes."""
     aioclient_mock.get(
         duckdns.UPDATE_URL, params={"domains": DOMAIN, "token": TOKEN}, text="OK"
     )
 
-    result = yield from async_setup_component(
+    result = await async_setup_component(
         hass, duckdns.DOMAIN, {"duckdns": {"domain": DOMAIN, "access_token": TOKEN}}
     )
+
+    await hass.async_block_till_done()
+
     assert result
     assert aioclient_mock.call_count == 1
 
     async_fire_time_changed(hass, utcnow() + timedelta(minutes=5))
-    yield from hass.async_block_till_done()
+    await hass.async_block_till_done()
     assert aioclient_mock.call_count == 2
 
 
-@asyncio.coroutine
-def test_setup_fails_if_update_fails(hass, aioclient_mock):
+async def test_setup_backoff(hass, aioclient_mock):
     """Test setup fails if first update fails."""
     aioclient_mock.get(
         duckdns.UPDATE_URL, params={"domains": DOMAIN, "token": TOKEN}, text="KO"
     )
 
-    result = yield from async_setup_component(
+    result = await async_setup_component(
         hass, duckdns.DOMAIN, {"duckdns": {"domain": DOMAIN, "access_token": TOKEN}}
     )
-    assert not result
+    assert result
+    await hass.async_block_till_done()
     assert aioclient_mock.call_count == 1
 
+    # Copy of the DuckDNS intervals from duckdns/__init__.py
+    intervals = (
+        INTERVAL,
+        timedelta(minutes=1),
+        timedelta(minutes=5),
+        timedelta(minutes=15),
+        timedelta(minutes=30),
+    )
+    tme = utcnow()
+    await hass.async_block_till_done()
 
-@asyncio.coroutine
-def test_service_set_txt(hass, aioclient_mock, setup_duckdns):
+    _LOGGER.debug("Backoff...")
+    for idx in range(1, len(intervals)):
+        tme += intervals[idx]
+        async_fire_time_changed(hass, tme)
+        await hass.async_block_till_done()
+
+        assert aioclient_mock.call_count == idx + 1
+
+
+async def test_service_set_txt(hass, aioclient_mock, setup_duckdns):
     """Test set txt service call."""
     # Empty the fixture mock requests
     aioclient_mock.clear_requests()
@@ -86,12 +107,11 @@ def test_service_set_txt(hass, aioclient_mock, setup_duckdns):
     )
 
     assert aioclient_mock.call_count == 0
-    yield from async_set_txt(hass, "some-txt")
+    await async_set_txt(hass, "some-txt")
     assert aioclient_mock.call_count == 1
 
 
-@asyncio.coroutine
-def test_service_clear_txt(hass, aioclient_mock, setup_duckdns):
+async def test_service_clear_txt(hass, aioclient_mock, setup_duckdns):
     """Test clear txt service call."""
     # Empty the fixture mock requests
     aioclient_mock.clear_requests()
@@ -103,5 +123,66 @@ def test_service_clear_txt(hass, aioclient_mock, setup_duckdns):
     )
 
     assert aioclient_mock.call_count == 0
-    yield from async_set_txt(hass, None)
+    await async_set_txt(hass, None)
     assert aioclient_mock.call_count == 1
+
+
+async def test_async_track_time_interval_backoff(hass):
+    """Test setup fails if first update fails."""
+    ret_val = False
+    call_count = 0
+    tme = None
+
+    async def _return(now):
+        nonlocal call_count, ret_val, tme
+        if tme is None:
+            tme = now
+        call_count += 1
+        return ret_val
+
+    intervals = (
+        INTERVAL,
+        INTERVAL * 2,
+        INTERVAL * 5,
+        INTERVAL * 9,
+        INTERVAL * 10,
+        INTERVAL * 11,
+        INTERVAL * 12,
+    )
+
+    async_track_time_interval_backoff(hass, _return, intervals)
+    await hass.async_block_till_done()
+
+    assert call_count == 1
+
+    _LOGGER.debug("Backoff...")
+    for idx in range(1, len(intervals)):
+        tme += intervals[idx]
+        async_fire_time_changed(hass, tme)
+        await hass.async_block_till_done()
+
+        assert call_count == idx + 1
+
+    _LOGGER.debug("Max backoff reached - intervals[-1]")
+    for _idx in range(1, 10):
+        tme += intervals[-1]
+        async_fire_time_changed(hass, tme)
+        await hass.async_block_till_done()
+
+        assert call_count == idx + 1 + _idx
+
+    _LOGGER.debug("Reset backoff")
+    call_count = 0
+    ret_val = True
+    tme += intervals[-1]
+    async_fire_time_changed(hass, tme)
+    await hass.async_block_till_done()
+    assert call_count == 1
+
+    _LOGGER.debug("No backoff - intervals[0]")
+    for _idx in range(2, 10):
+        tme += intervals[0]
+        async_fire_time_changed(hass, tme)
+        await hass.async_block_till_done()
+
+        assert call_count == _idx