Merge lp:~al-maisan/landscape-client/obsolete-results-v into lp:~landscape/landscape-client/trunk

[Muharem Hrnjadovic]

Obsolete response messages will now be discarded as opposed to being sent to the server.

A response message is considered obsolete if the secure ID has changed since the request message was received.

[lp:~al-maisan/landscape-client/obsolete-results-v]

244. By Muharem Hrnjadovic on 2010-04-29

bzr ls-lint fixes

245. By Muharem Hrnjadovic on 2010-04-29

Simplified test.

246. By Muharem Hrnjadovic on 2010-04-29

Fixed doc string.

247. By Muharem Hrnjadovic on 2010-04-29

Simplified obsolete response logic.

[Free Ekanayaka (free.ekanayaka)]

Very nice work, just a few comments but overall the branch looks great to me.

[1]

+ def __init__(self, db, id):

Please add @param entries in the class docstring for these two parameters.

[2]

+ @param filename: The file that contains the sqlite database.

Jamu recently pointed that @param docstrings for __init__ method should be actually put in the class docstring, like:

class ExchangeStore(object):
    """Message meta data required by the L{MessageExchange}.

    The implementation uses a SQLite database as backend, with a single table
    called "message_context", whose schema is defined in
    L{ensure_exchange_schema}.

    @param filename: The file that contains the sqlite database.
    """

because the callable that gets actually used is the class itself (you usually don't call __init__ directly).

[3]

For consistency with other tests and for easier reading please name test cases against the tests the exercise, plus some test specific suffix. A few examples:

test_add_operationid_is_unique -> test_add_message_context_with_duplicate_operation_id
test_get_message_context_works -> test_get_message_context
test_deleting_message_contexts_works -> test_deleting_message_contexts

[4]

+ self._store = ExchangeStore(

It would be better to use dependency injection (that is usually the pattern adopted in the client) and pass an instance of ExchangeStore instead of creating it in the __init__ method. This helps testing and mocking, and make it easier in general to pass objects of other types providing the same API.

For example your test case will then have a self.exchange_store attribute (that you use in to build the MessageExchange object under test) which you can use for performing assertions instead of accessing self.exchanger._store, which is a private attribute and should be considered an implementation detail.

[5]

+ return 0

Maybe None would be safer, as 0 could be used as message id number.

[6]

+ self.assertIsNot(
+ None,
+ self.exchanger._store.get_message_context(message['operation-id']))

I think the test should also check that the created MessageContext has the expected attributes set:

+ message_context = self.exchanger._store.get_message_context(message['operation-id'])
+ self.assertEquals(message_context.operation_id, 123456)
+ self.assertEquals(message_context.message_type, "type-R")

[7]

+ self.exchanger.send({"type": "resynchronize", "operation-id": 234567})

Assuming that send() returns None in this case, the test should assert it:

+ message_id = self.exchanger.send({"type": "resynchronize", "operation-id": 234567})
+ self.assertIs(message_id, None)

[8]

+ logging.info(
+ "Response message with operation-id %s was discarded "
+ "because the client's secure ID has changed in the meantime"
+ % message.get('operation-id'))

The logging is not tested, you can use self.logfile.getvalue() for that.

review: Needs Fixing

[lp:~al-maisan/landscape-client/obsolete-results-v]

248. By Muharem Hrnjadovic on 2010-04-29

Free's review comments [1-5]

249. By Muharem Hrnjadovic on 2010-04-29

Free's review comment #6

250. By Muharem Hrnjadovic on 2010-04-29

Free's review comment #7

251. By Muharem Hrnjadovic on 2010-04-29

Free's review comment #7

252. By Muharem Hrnjadovic on 2010-04-29

Free's review comment #8

[Thomas Herve (therve)]

It looks good!

[1] As it, the message context store will grow for ever. I think you should remove messages from it as soon as you add it to the message store (or discard it).

[2]
+ def _message_is_obsolete(self, message):
+ """True if message is obsolete.

Format "Returns C{True} if message is obsolete."

[3]
+class MessageContext(object):
+ """Stores the secure ID for incoming messages that require a response.

It seems to store more than that now.

[4]
+ """Accessing a C{MessageContext} with an existing
+ C{operation-id} works."""

+ """Calling C{all_operation_ids} on an empty database returns an empty
+ list."""

Please put the docstring on their own lines.

Thanks!

review: Needs Fixing

[lp:~al-maisan/landscape-client/obsolete-results-v]

253. By Muharem Hrnjadovic on 2010-04-30

Thomas' review comment #1

254. By Muharem Hrnjadovic on 2010-04-30

Thomas' review comment #2

255. By Muharem Hrnjadovic on 2010-04-30

Thomas' review comment #3

256. By Muharem Hrnjadovic on 2010-04-30

Thomas' review comment #4

[Muharem Hrnjadovic (al-maisan)]

On 04/29/2010 05:31 PM, Free Ekanayaka wrote:
> Review: Needs Fixing
Hello Free,

I addressed all of your comments below, please review the changes. An
incremental diff is attached to this message for your convenience.

> Very nice work, just a few comments but overall the branch looks great to me.
>
> [1]
>
> + def __init__(self, db, id):
>
> Please add @param entries in the class docstring for these two parameters.
>
> [2]
>
> + @param filename: The file that contains the sqlite database.
>
> Jamu recently pointed that @param docstrings for __init__ method should be actually put in the class docstring, like:
>
> class ExchangeStore(object):
> """Message meta data required by the L{MessageExchange}.
>
> The implementation uses a SQLite database as backend, with a single table
> called "message_context", whose schema is defined in
> L{ensure_exchange_schema}.
>
> @param filename: The file that contains the sqlite database.
> """
>
> because the callable that gets actually used is the class itself (you usually don't call __init__ directly).
>
> [3]
>
> For consistency with other tests and for easier reading please name test cases against the tests the exercise, plus some test specific suffix. A few examples:
>
> test_add_operationid_is_unique -> test_add_message_context_with_duplicate_operation_id
> test_get_message_context_works -> test_get_message_context
> test_deleting_message_contexts_works -> test_deleting_message_contexts
>
> [4]
>
> + self._store = ExchangeStore(
>
> It would be better to use dependency injection (that is usually the pattern adopted in the client) and pass an instance of ExchangeStore instead of creating it in the __init__ method. This helps testing and mocking, and make it easier in general to pass objects of other types providing the same API.
>
> For example your test case will then have a self.exchange_store attribute (that you use in to build the MessageExchange object under test) which you can use for performing assertions instead of accessing self.exchanger._store, which is a private attribute and should be considered an implementation detail.
>
> [5]
>
> + return 0
>
> Maybe None would be safer, as 0 could be used as message id number.
>
> [6]
>
> + self.assertIsNot(
> + None,
> + self.exchanger._store.get_message_context(message['operation-id']))
>
> I think the test should also check that the created MessageContext has the expected attributes set:
>
> + message_context = self.exchanger._store.get_message_context(message['operation-id'])
> + self.assertEquals(message_context.operation_id, 123456)
> + self.assertEquals(message_context.message_type, "type-R")
>
> [7]
>
> + self.exchanger.send({"type": "resynchronize", "operation-id": 234567})
>
> Assuming that send() returns None in this case, the test should assert it:
>
> + message_id = self.exchanger.send({"type": "resynchronize", "operation-id": 234567})
> + self.assertIs(message_id, None)
>
> [8]
>
> + logging.info(
> + "Response message with operation-id %s was discarded "
> + ...

[Muharem Hrnjadovic (al-maisan)]

On 04/30/2010 09:54 AM, Thomas Herve wrote:
> Review: Needs Fixing
> It looks good!
Hello Thomas,

thank you very much for the review and especially for the first catch below.
I have addressed all of your comments below, please review the changes. An
incremental diff is attached to this message for your convenience.

> [1] As it, the message context store will grow for ever. I think you should remove messages from it as soon as you add it to the message store (or discard it).
>
> [2]
> + def _message_is_obsolete(self, message):
> + """True if message is obsolete.
>
> Format "Returns C{True} if message is obsolete."
>
> [3]
> +class MessageContext(object):
> + """Stores the secure ID for incoming messages that require a response.
>
> It seems to store more than that now.
>
> [4]
> + """Accessing a C{MessageContext} with an existing
> + C{operation-id} works."""
>
>
> + """Calling C{all_operation_ids} on an empty database returns an empty
> + list."""
>
> Please put the docstring on their own lines.
>
>
> Thanks!

Best regards

--
Muharem Hrnjadovic <email address hidden>
Public key id : B2BBFCFC
Key fingerprint : A5A3 CC67 2B87 D641 103F 5602 219F 6B60 B2BB FCFC

[Free Ekanayaka (free.ekanayaka)]

[9]

+ message_context_store,

Please name this parameter "exchange_store", so it matches the name of the expected type (ExchangeStore) and it is shorter. Also, now that we have it, the ExchangeStore could be used to store also other types of information beside MessageContext's, so it makes sense to use a generic name for it.

+ self._store = message_context_store

Similarly please name this attribute "self._exchange_store", to make it clear which store is referring to (there is another attribute called "self._message_store", so the naming will be consistent).

[10]

+ else:
+ if "timestamp" not in message:
+ message["timestamp"] = int(self._reactor.time())
+ message_id = self._message_store.add(message)
+ if urgent:
+ self.schedule_exchange(urgent=True)
+ return message_id

I think you can save this else/indentation level and simply write:

+ if "timestamp" not in message:
+ message["timestamp"] = int(self._reactor.time())
+ message_id = self._message_store.add(message)
+ if urgent:
+ self.schedule_exchange(urgent=True)
+ return message_id

I personally like the pattern when you check a series of early-return conditions, before reaching the meat of a function/method, like

if this:
   return something
if that:
   return something_else

do_the_actual_thing

[11]

+ exchange_store = ExchangeStore(
+ os.path.join(config.data_path, "exchange.database"))

Just to avoid hard-coding this value you could add a property to l.broker.config.BrokerConfiguration, like:

@property
def exchange_store_path(self):
    return os.path.join(self.data_path, "exchange.database"))

and use it as

+ exchange_store = ExchangeStore(self.config.exchange_store_path)

[12]

+ exchange_store = ExchangeStore(
+ os.path.join(test_case.config.data_path, "exchange.database"))

Please add this object to the test_case attributes, so it can be used in tests:

+ test_case.exchange_store = ExchangeStore(
+ os.path.join(test_case.config.data_path, "exchange.database"))

after that you can avoid creating it explicitly in all the tests.

[13]

+ self.exchanger._store.get_message_context(message['operation-id']))

Once you have self.exchange_store as test attribute (see [12]) you should use it instead of accessing this private attribute.

Thanks!

[Muharem Hrnjadovic (al-maisan)]

Hello Free,

thank you very much for these comments, I really appreciate it.
I have taken care of all of them, please see the attached incremental
diff.

On 04/30/2010 11:14 AM, Free Ekanayaka wrote:
> [9]
>
> + message_context_store,
>
> Please name this parameter "exchange_store", so it matches the name of the expected type (ExchangeStore) and it is shorter. Also, now that we have it, the ExchangeStore could be used to store also other types of information beside MessageContext's, so it makes sense to use a generic name for it.
>
> + self._store = message_context_store
>
> Similarly please name this attribute "self._exchange_store", to make it clear which store is referring to (there is another attribute called "self._message_store", so the naming will be consistent).
>
> [10]
>
> + else:
> + if "timestamp" not in message:
> + message["timestamp"] = int(self._reactor.time())
> + message_id = self._message_store.add(message)
> + if urgent:
> + self.schedule_exchange(urgent=True)
> + return message_id
>
> I think you can save this else/indentation level and simply write:
>
> + if "timestamp" not in message:
> + message["timestamp"] = int(self._reactor.time())
> + message_id = self._message_store.add(message)
> + if urgent:
> + self.schedule_exchange(urgent=True)
> + return message_id
>
> I personally like the pattern when you check a series of early-return conditions, before reaching the meat of a function/method, like
>
> if this:
> return something
> if that:
> return something_else
>
> do_the_actual_thing
>
> [11]
>
> + exchange_store = ExchangeStore(
> + os.path.join(config.data_path, "exchange.database"))
>
> Just to avoid hard-coding this value you could add a property to l.broker.config.BrokerConfiguration, like:
>
> @property
> def exchange_store_path(self):
> return os.path.join(self.data_path, "exchange.database"))
>
> and use it as
>
> + exchange_store = ExchangeStore(self.config.exchange_store_path)
>
> [12]
>
> + exchange_store = ExchangeStore(
> + os.path.join(test_case.config.data_path, "exchange.database"))
>
> Please add this object to the test_case attributes, so it can be used in tests:
>
> + test_case.exchange_store = ExchangeStore(
> + os.path.join(test_case.config.data_path, "exchange.database"))
>
> after that you can avoid creating it explicitly in all the tests.
>
> [13]
>
> + self.exchanger._store.get_message_context(message['operation-id']))
>
> Once you have self.exchange_store as test attribute (see [12]) you should use it instead of accessing this private attribute.
>
> Thanks!

Best regards

--
Muharem Hrnjadovic <email address hidden>
Public key id : B2BBFCFC
Key fingerprint : A5A3 CC67 2B87 D641 103F 5602 219F 6B60 B2BB FCFC

[lp:~al-maisan/landscape-client/obsolete-results-v]

257. By Muharem Hrnjadovic on 2010-04-30

Free's review comments [9,10]

258. By Muharem Hrnjadovic on 2010-04-30

Free's review comment #11

259. By Muharem Hrnjadovic on 2010-04-30

Free's review comment #12

260. By Muharem Hrnjadovic on 2010-04-30

Free's review comments [12,13]

[Free Ekanayaka (free.ekanayaka)]

The branch looks great! Thanks for addressing all the comments. +1

[14]

+ test_case.exchange_store = ExchangeStore(
+ os.path.join(test_case.config.data_path, "exchange.database"))

You can now use BrokerConfiguration.exchange_store_path:

+ test_case.exchange_store = ExchangeStore(
+ test_case.config.exchange_store_path)

review: Approve

[lp:~al-maisan/landscape-client/obsolete-results-v]

261. By Muharem Hrnjadovic on 2010-04-30

Free's review comment #14

[Thomas Herve (therve)]

[5] Some pyflakes:
landscape/broker/tests/test_exchange.py:1: 'os' imported but unused
landscape/broker/tests/test_exchange.py:8: 'ExchangeStore' imported but unused

[6] get_message_context should retrieve all the data from the database in one query, and pass it to the MessageContext __init__, thus removing the select in there.

+1!

review: Approve

[lp:~al-maisan/landscape-client/obsolete-results-v]

262. By Muharem Hrnjadovic on 2010-04-30

Thomas' review comment #5

[Muharem Hrnjadovic (al-maisan)]

On 04/30/2010 04:59 PM, Thomas Herve wrote:
> Review: Approve
> [5] Some pyflakes:
> landscape/broker/tests/test_exchange.py:1: 'os' imported but unused
> landscape/broker/tests/test_exchange.py:8: 'ExchangeStore' imported but unused
>
> [6] get_message_context should retrieve all the data from the database in one query, and pass it to the MessageContext __init__, thus removing the select in there.

Hello Thomas,

thanks for the suggestions above! I have revised the code accordingly.
Please see the attached incremental diff.

Best regards

--
Muharem Hrnjadovic <email address hidden>
Public key id : B2BBFCFC
Key fingerprint : A5A3 CC67 2B87 D641 103F 5602 219F 6B60 B2BB FCFC

[lp:~al-maisan/landscape-client/obsolete-results-v]

263. By Muharem Hrnjadovic on 2010-04-30

Thomas' review comment #6

264. By Muharem Hrnjadovic on 2010-04-30

Made code more succinct

=== modified file 'landscape/broker/config.py'
--- landscape/broker/config.py	2010-01-06 15:11:24 +0000
+++ landscape/broker/config.py	2010-04-30 19:04:26 +0000
@@ -18,6 +18,10 @@
         self._original_http_proxy = os.environ.get("http_proxy")
         self._original_https_proxy = os.environ.get("https_proxy")
 
+    @property
+    def exchange_store_path(self):
+        return os.path.join(self.data_path, "exchange.database")
+
     def make_parser(self):
         """Parser factory for broker-specific options.
 
 
=== modified file 'landscape/broker/exchange.py'
--- landscape/broker/exchange.py	2010-01-22 11:51:19 +0000
+++ landscape/broker/exchange.py	2010-04-30 19:04:26 +0000
@@ -22,6 +22,7 @@
     plugin_name = "message-exchange"
 
     def __init__(self, reactor, store, transport, registration_info,
+                 exchange_store,
                  exchange_interval=60*60,
                  urgent_exchange_interval=10,
                  monitor_interval=None,
@@ -52,6 +53,7 @@
         self._client_accepted_types = set()
         self._client_accepted_types_hash = None
         self._message_handlers = {}
+        self._exchange_store = exchange_store
 
         self.register_message("accepted-types", self._handle_accepted_types)
         self.register_message("resynchronize", self._handle_resynchronize)
@@ -63,6 +65,30 @@
         """Return a binary tuple with urgent and normal exchange intervals."""
         return (self._urgent_exchange_interval, self._exchange_interval)
 
+    def _message_is_obsolete(self, message):
+        """Returns C{True} if message is obsolete.
+
+        A message is considered obsolete if the secure ID changed since it was
+        received.
+        """
+        if 'operation-id' not in message:
+            return False
+
+        operation_id = message['operation-id']
+        context = self._exchange_store.get_message_context(operation_id)
+        if context is None:
+            logging.warning(
+                "No message context for message with operation-id: %s"
+                % operation_id)
+            return False
+
+        # Compare the current secure ID with the one that was in effect when
+        # the request message was received.
+        result = self._registration_info.secure_id != context.secure_id
+        context.remove()
+
+        return result
+
     def send(self, message, urgent=False):
         """Include a message to be sent in an exchange.
 
@@ -71,6 +97,13 @@
 
         @param message: Same as in L{MessageStore.add}.
         """
+        if self._message_is_obsolete(message):
+            logging.info(
+                "Response message with operation-id %s was discarded "
+                "because the client's secure ID has changed in the meantime"
+                % message.get('operation-id'))
+            return None
+
         if "timestamp" not in message:
             message["timestamp"] = int(self._reactor.time())
         message_id = self._message_store.add(message)
@@ -300,7 +333,8 @@
         @param result: The response got in reply to the C{payload}.
         """
         message_store = self._message_store
-        self._client_accepted_types_hash = result.get("client-accepted-types-hash")
+        self._client_accepted_types_hash = result.get(
+            "client-accepted-types-hash")
         next_expected = result.get("next-expected-sequence")
         old_sequence = message_store.get_sequence()
         if next_expected is None:
@@ -363,6 +397,13 @@
         Any message handlers registered with L{register_message} will
         be called.
         """
+        if 'operation-id' in message:
+            # This is a message that requires a response. Store the secure ID
+            # so we can check for obsolete results later.
+            self._exchange_store.add_message_context(
+                message['operation-id'], self._registration_info.secure_id,
+                message['type'])
+
         self._reactor.fire("message", message)
         # This has plan interference! but whatever.
         if message["type"] in self._message_handlers:
 
=== added file 'landscape/broker/exchangestore.py'
--- landscape/broker/exchangestore.py	1970-01-01 00:00:00 +0000
+++ landscape/broker/exchangestore.py	2010-04-30 19:04:26 +0000
@@ -0,0 +1,110 @@
+"""Provide access to the persistent data used by the L{MessageExchange}."""
+import time
+
+try:
+    import sqlite3
+except ImportError:
+    from pysqlite2 import dbapi2 as sqlite3
+
+from landscape.lib.store import with_cursor
+
+
+class MessageContext(object):
+    """Stores a context for incoming messages that require a response.
+
+    The context consists of
+
+      - the "operation-id" value
+      - the secure ID that was in effect when the message was received
+      - the message type
+      - the time when the message was received
+
+    This data will be used to detect secure ID changes between the time at
+    which the request message came in and the completion of the request.
+    If the secure ID did change the result message is obolete and will not be
+    sent to the server.
+
+    @param db: the sqlite database handle.
+    @param id: the database key value for this instance.
+    """
+
+    def __init__(self, db, operation_id, secure_id, message_type, timestamp):
+        self._db = db
+        self.operation_id = operation_id
+        self.secure_id = secure_id
+        self.message_type = message_type
+        self.timestamp = timestamp
+
+    @with_cursor
+    def remove(self, cursor):
+        cursor.execute(
+            "DELETE FROM message_context WHERE operation_id=?",
+            (self.operation_id,))
+
+
+class ExchangeStore(object):
+    """Message meta data required by the L{MessageExchange}.
+
+    The implementation uses a SQLite database as backend, with a single table
+    called "message_context", whose schema is defined in
+    L{ensure_exchange_schema}.
+
+    @param filename: The name of the file that contains the sqlite database.
+    """
+    _db = None
+
+    def __init__(self, filename):
+        self._filename = filename
+
+    def _ensure_schema(self):
+        ensure_exchange_schema(self._db)
+
+    @with_cursor
+    def add_message_context(
+        self, cursor, operation_id, secure_id, message_type):
+        """Add a L{MessageContext} with the given data."""
+        params = (operation_id, secure_id, message_type, time.time())
+        cursor.execute(
+            "INSERT INTO message_context "
+            "   (operation_id, secure_id, message_type, timestamp) "
+            "   VALUES (?,?,?,?)", params)
+        return MessageContext(self._db, *params)
+
+    @with_cursor
+    def get_message_context(self, cursor, operation_id):
+        """The L{MessageContext} for the given C{operation_id} or C{None}."""
+        cursor.execute(
+            "SELECT operation_id, secure_id, message_type, timestamp "
+            "FROM message_context WHERE operation_id=?", (operation_id,))
+        row = cursor.fetchone()
+        return MessageContext(self._db, *row) if row else None
+
+    @with_cursor
+    def all_operation_ids(self, cursor):
+        """Return all operation IDs currently stored in C{message_context}."""
+        cursor.execute("SELECT operation_id FROM message_context")
+        result = cursor.fetchall()
+        return [row[0] for row in result]
+
+
+def ensure_exchange_schema(db):
+    """Create all tables needed by a L{ExchangeStore}.
+
+    @param db: A connection to a SQLite database.
+    """
+    cursor = db.cursor()
+    try:
+        cursor.execute(
+            "CREATE TABLE message_context"
+            " (id INTEGER PRIMARY KEY, timestamp TIMESTAMP, "
+            "  secure_id TEXT NOT NULL, operation_id INTEGER NOT NULL, "
+            "  message_type text NOT NULL)")
+        cursor.execute(
+            "CREATE UNIQUE INDEX msgctx_operationid_idx ON "
+            "message_context(operation_id)")
+    except (sqlite3.OperationalError, sqlite3.DatabaseError):
+        cursor.close()
+        db.rollback()
+    else:
+        cursor.close()
+        db.commit()
 
=== modified file 'landscape/broker/service.py'
--- landscape/broker/service.py	2010-04-23 11:37:25 +0000
+++ landscape/broker/service.py	2010-04-30 19:04:26 +0000
@@ -7,6 +7,7 @@
 from landscape.broker.config import BrokerConfiguration
 from landscape.broker.transport import HTTPTransport
 from landscape.broker.exchange import MessageExchange
+from landscape.broker.exchangestore import ExchangeStore
 from landscape.broker.ping import Pinger
 from landscape.broker.store import get_default_message_store
 from landscape.broker.server import BrokerServer
@@ -52,9 +53,11 @@
         self.message_store = get_default_message_store(
             self.persist, config.message_store_path)
         self.identity = Identity(self.config, self.persist)
+        exchange_store = ExchangeStore(self.config.exchange_store_path)
         self.exchanger = MessageExchange(
             self.reactor, self.message_store, self.transport, self.identity,
-            config.exchange_interval, config.urgent_exchange_interval)
+            exchange_store, config.exchange_interval,
+            config.urgent_exchange_interval)
         self.pinger = self.pinger_factory(self.reactor, config.ping_url,
                                           self.identity, self.exchanger)
         self.registration = RegistrationHandler(
 
=== modified file 'landscape/broker/tests/helpers.py'
--- landscape/broker/tests/helpers.py	2010-04-07 21:00:46 +0000
+++ landscape/broker/tests/helpers.py	2010-04-30 19:04:26 +0000
@@ -6,6 +6,7 @@
 from landscape.reactor import FakeReactor
 from landscape.broker.transport import FakeTransport
 from landscape.broker.exchange import MessageExchange
+from landscape.broker.exchangestore import ExchangeStore
 from landscape.broker.store import get_default_message_store
 from landscape.broker.registration import Identity, RegistrationHandler
 from landscape.broker.ping import Pinger
@@ -74,9 +75,12 @@
         test_case.transport = FakeTransport(test_case.config.url,
                                             test_case.config.ssl_public_key)
         test_case.reactor = FakeReactor()
+        test_case.exchange_store = ExchangeStore(
+            test_case.config.exchange_store_path)
         test_case.exchanger = MessageExchange(
             test_case.reactor, test_case.mstore, test_case.transport,
-            test_case.identity, test_case.config.exchange_interval,
+            test_case.identity, test_case.exchange_store,
+            test_case.config.exchange_interval,
             test_case.config.urgent_exchange_interval)
 
 
 
=== modified file 'landscape/broker/tests/test_exchange.py'
--- landscape/broker/tests/test_exchange.py	2010-04-28 13:07:53 +0000
+++ landscape/broker/tests/test_exchange.py	2010-04-30 19:04:26 +0000
@@ -1,4 +1,3 @@
-
 from landscape import SERVER_API, CLIENT_API
 from landscape.lib.persist import Persist
 from landscape.lib.hashlib import md5
@@ -22,6 +21,7 @@
         self.mstore.add_schema(Message("empty", {}))
         self.mstore.add_schema(Message("data", {"data": Int()}))
         self.mstore.add_schema(Message("holdme", {}))
+        self.identity.secure_id = 'needs-to-be-set-for-tests-to-pass'
 
     def wait_for_exchange(self, urgent=False, factor=1, delta=0):
         if urgent:
@@ -240,7 +240,6 @@
         self.mstore.add({"type": "data", "data": 2})
         self.mstore.add({"type": "data", "data": 3})
         # next one, server will respond with 1!
-
         def desynched_send_data(payload, computer_id=None, message_api=None):
             self.transport.next_expected_sequence = 1
             return {"next-expected-sequence": 1}
@@ -273,7 +272,7 @@
         """
         transport = FakeTransport()
         exchanger = MessageExchange(self.reactor, self.mstore, transport,
-                                    self.identity)
+                                    self.identity, self.exchange_store)
         exchanger.start()
         self.wait_for_exchange(urgent=True)
         self.assertEquals(len(transport.payloads), 1)
@@ -390,6 +389,7 @@
         del self.transport.exchange
 
         exchanged = []
+
         def exchange_callback():
             exchanged.append(True)
 
@@ -473,7 +473,6 @@
         self.assertEquals(payload.get("server-api"), "2.0")
         self.assertEquals(self.transport.message_api, "2.0")
 
-
     def test_include_total_messages_none(self):
         """
         The payload includes the total number of messages that the client has
@@ -500,14 +499,14 @@
         pending.
         """
         exchanger = MessageExchange(self.reactor, self.mstore, self.transport,
-                                    self.identity, max_messages=1)
+                                    self.identity, self.exchange_store,
+                                    max_messages=1)
         self.mstore.set_accepted_types(["empty"])
         self.mstore.add({"type": "empty"})
         self.mstore.add({"type": "empty"})
         exchanger.exchange()
         self.assertEquals(self.transport.payloads[0]["total-messages"], 2)
 
-
     def test_impending_exchange(self):
         """
         A reactor event is emitted shortly (10 seconds) before an exchange
@@ -530,7 +529,8 @@
         # fixture has an urgent exchange interval of 10 seconds, which makes
         # testing this awkward.
         exchanger = MessageExchange(self.reactor, self.mstore, self.transport,
-                                    self.identity, urgent_exchange_interval=20)
+                                    self.identity, self.exchange_store,
+                                    urgent_exchange_interval=20)
         exchanger.schedule_exchange(urgent=True)
         events = []
         self.reactor.call_on("impending-exchange", lambda: events.append(True))
@@ -547,7 +547,8 @@
         before the new urgent exchange.
         """
         exchanger = MessageExchange(self.reactor, self.mstore, self.transport,
-                                    self.identity, urgent_exchange_interval=20)
+                                    self.identity, self.exchange_store,
+                                    urgent_exchange_interval=20)
         events = []
         self.reactor.call_on("impending-exchange", lambda: events.append(True))
         # This call will:
@@ -610,6 +611,7 @@
         self.transport.exchange = failed_send_data
 
         exchanged = []
+
         def exchange_failed_callback():
             exchanged.append(True)
 
@@ -700,6 +702,7 @@
 
         def handler1(message):
             messages.append(("one", message))
+
         def handler2(message):
             messages.append(("two", message))
 
@@ -722,6 +725,7 @@
 
     def test_server_uuid_change_cause_event(self):
         called = []
+
         def server_uuid_changed(old_uuid, new_uuid):
             called.append((old_uuid, new_uuid))
         self.reactor.call_on("server-uuid-changed", server_uuid_changed)
@@ -756,6 +760,7 @@
         the event is not emitted.
         """
         called = []
+
         def server_uuid_changed(old_uuid, new_uuid):
             called.append((old_uuid, new_uuid))
         self.reactor.call_on("server-uuid-changed", server_uuid_changed)
@@ -779,6 +784,77 @@
 
         self.assertNotIn("INFO: Server UUID changed", self.logfile.getvalue())
 
+    def test_return_messages_have_their_context_stored(self):
+        """
+        Incoming messages with an 'operation-id' key will have the secure id
+        stored in the L{ExchangeStore}.
+        """
+        messages = []
+        self.exchanger.register_message("type-R", messages.append)
+        msg = {"type": "type-R", "whatever": 5678, "operation-id": 123456}
+        server_message = [msg]
+        self.transport.responses.append(server_message)
+        self.exchanger.exchange()
+        [message] = messages
+        self.assertIsNot(
+            None,
+            self.exchange_store.get_message_context(message['operation-id']))
+        message_context = self.exchange_store.get_message_context(
+            message['operation-id'])
+        self.assertEquals(message_context.operation_id, 123456)
+        self.assertEquals(message_context.message_type, "type-R")
+
+    def test_one_way_messages_do_not_have_their_context_stored(self):
+        """
+        Incoming messages without an 'operation-id' key will *not* have the
+        secure id stored in the L{ExchangeStore}.
+        """
+        ids_before = self.exchange_store.all_operation_ids()
+
+        msg = {"type": "type-R", "whatever": 5678}
+        server_message = [msg]
+        self.transport.responses.append(server_message)
+        self.exchanger.exchange()
+
+        ids_after = self.exchange_store.all_operation_ids()
+        self.assertEquals(ids_before, ids_after)
+
+    def test_obsolete_response_messages_are_discarded(self):
+        """
+        An obsolete response message will be discarded as opposed to being
+        sent to the server.
+
+        A response message is considered obsolete if the secure ID changed
+        since the request message was received.
+        """
+        # Receive the message below from the server.
+        msg = {"type": "type-R", "whatever": 5678, "operation-id": 234567}
+        server_message = [msg]
+        self.transport.responses.append(server_message)
+        self.exchanger.exchange()
+
+        # Change the secure ID so that the response message gets discarded.
+        self.identity.secure_id = 'brand-new'
+        ids_before = self.exchange_store.all_operation_ids()
+
+        self.mstore.set_accepted_types(["resynchronize"])
+        message_id = self.exchanger.send(
+            {"type": "resynchronize", "operation-id": 234567})
+        self.exchanger.exchange()
+        self.assertEquals(2, len(self.transport.payloads))
+        messages = self.transport.payloads[1]["messages"]
+        self.assertEquals([], messages)
+        self.assertIs(None, message_id)
+        expected_log_entry = (
+            "Response message with operation-id 234567 was discarded because "
+            "the client's secure ID has changed in the meantime")
+        self.assertTrue(expected_log_entry in self.logfile.getvalue())
+
+        # The MessageContext was removed after utilisation.
+        ids_after = self.exchange_store.all_operation_ids()
+        self.assertTrue(len(ids_after) == len(ids_before) - 1)
+        self.assertFalse('234567' in ids_after)
+
 
 class AcceptedTypesMessageExchangeTest(LandscapeTest):
 
@@ -868,7 +944,6 @@
             self.transport.payloads[2]["client-accepted-types"],
             sorted(["type-A"] + DEFAULT_ACCEPTED_TYPES))
 
-
     def test_register_message_adds_accepted_type(self):
         """
         Using the C{register_message} method of the exchanger causes
 
=== added file 'landscape/broker/tests/test_exchangestore.py'
--- landscape/broker/tests/test_exchangestore.py	1970-01-01 00:00:00 +0000
+++ landscape/broker/tests/test_exchangestore.py	2010-04-30 19:04:26 +0000
@@ -0,0 +1,84 @@
+import time
+
+try:
+    import sqlite3
+except ImportError:
+    from pysqlite2 import dbapi2 as sqlite3
+
+from landscape.tests.helpers import LandscapeTest
+
+from landscape.broker.exchangestore import ExchangeStore
+
+
+class ExchangeStoreTest(LandscapeTest):
+    """Unit tests for the C{ExchangeStore}."""
+
+    def setUp(self):
+        super(ExchangeStoreTest, self).setUp()
+
+        self.filename = self.makeFile()
+        self.store1 = ExchangeStore(self.filename)
+        self.store2 = ExchangeStore(self.filename)
+
+    def test_add_message_context(self):
+        """Adding a message context works correctly."""
+        now = time.time()
+        self.store1.add_message_context(123, 'abc', 'change-packages')
+
+        db = sqlite3.connect(self.store2._filename)
+        cursor = db.cursor()
+        cursor.execute(
+            "SELECT operation_id, secure_id, message_type, timestamp "
+            "FROM message_context WHERE operation_id=?", (123,))
+        results = cursor.fetchall()
+        self.assertEquals(1, len(results))
+        [row] = results
+        self.assertEquals(123, row[0])
+        self.assertEquals('abc', row[1])
+        self.assertEquals('change-packages', row[2])
+        self.assertTrue(row[3] > now)
+
+    def test_add_message_context_with_duplicate_operation_id(self):
+        """Only one message context with a given operation-id is permitted."""
+        self.store1.add_message_context(123, 'abc', 'change-packages')
+        self.assertRaises(
+            sqlite3.IntegrityError,
+            self.store1.add_message_context, 123, 'def', 'change-packages')
+
+    def test_get_message_context(self):
+        """
+        Accessing a C{MessageContext} with an existing C{operation-id} works.
+        """
+        now = time.time()
+        self.store1.add_message_context(234, 'bcd', 'change-packages')
+        context = self.store2.get_message_context(234)
+        self.assertEquals(234, context.operation_id)
+        self.assertEquals('bcd', context.secure_id)
+        self.assertEquals('change-packages', context.message_type)
+        self.assertTrue(context.timestamp > now)
+
+    def test_get_message_context_with_nonexistent_operation_id(self):
+        """Attempts to access a C{MessageContext} with a non-existent
+        C{operation-id} result in C{None}."""
+        self.assertIs(None, self.store1.get_message_context(999))
+
+    def test_message_context_remove(self):
+        """C{MessageContext}s are deleted correctly."""
+        context = self.store1.add_message_context(
+            345, 'opq', 'change-packages')
+        context.remove()
+        self.assertIs(None, self.store1.get_message_context(345))
+
+    def test_all_operation_ids_for_empty_database(self):
+        """
+        Calling C{all_operation_ids} on an empty database returns an empty
+        list.
+        """
+        self.assertEquals([], self.store1.all_operation_ids())
+
+    def test_all_operation_ids(self):
+        """C{all_operation_ids} works correctly."""
+        self.store1.add_message_context(456, 'cde', 'change-packages')
+        self.assertEquals([456], self.store2.all_operation_ids())
+        self.store2.add_message_context(567, 'def', 'change-packages')
+        self.assertEquals([456, 567], self.store1.all_operation_ids())
 
=== added file 'landscape/lib/store.py'
--- landscape/lib/store.py	1970-01-01 00:00:00 +0000
+++ landscape/lib/store.py	2010-04-30 19:04:26 +0000
@@ -0,0 +1,39 @@
+"""Functions used by all sqlite-backed stores."""
+
+try:
+    import sqlite3
+except ImportError:
+    from pysqlite2 import dbapi2 as sqlite3
+
+
+def with_cursor(method):
+    """Decorator that encloses the method in a database transaction.
+
+    Even though SQLite is supposed to be useful in autocommit mode, we've
+    found cases where the database continued to be locked for writing
+    until the cursor was closed.  With this in mind, instead of using
+    the autocommit mode, we explicitly terminate transactions and enforce
+    cursor closing with this decorator.
+    """
+
+    def inner(self, *args, **kwargs):
+        if not self._db:
+            # Create the database connection only when we start to actually
+            # use it. This is essentially just a workaroud of a sqlite bug
+            # happening when 2 concurrent processes try to create the tables
+            # around the same time, the one which fails having an incorrect
+            # cache and not seeing the tables
+            self._db = sqlite3.connect(self._filename)
+            self._ensure_schema()
+        try:
+            cursor = self._db.cursor()
+            try:
+                result = method(self, cursor, *args, **kwargs)
+            finally:
+                cursor.close()
+            self._db.commit()
+        except:
+            self._db.rollback()
+            raise
+        return result
+    return inner
 
=== modified file 'landscape/package/store.py'
--- landscape/package/store.py	2009-11-25 09:48:00 +0000
+++ landscape/package/store.py	2010-04-30 19:04:26 +0000
@@ -7,6 +7,7 @@
     from pysqlite2 import dbapi2 as sqlite3
 
 from landscape.lib import bpickle
+from landscape.lib.store import with_cursor
 
 
 class UnknownHashIDRequest(Exception):
@@ -17,39 +18,6 @@
     """Raised when trying to add an invalid hash=>id lookaside database."""
 
 
-def with_cursor(method):
-    """Decorator that encloses the method in a database transaction.
-
-    Even though SQLite is supposed to be useful in autocommit mode, we've
-    found cases where the database continued to be locked for writing
-    until the cursor was closed.  With this in mind, instead of using
-    the autocommit mode, we explicitly terminate transactions and enforce
-    cursor closing with this decorator.
-    """
-
-    def inner(self, *args, **kwargs):
-        if not self._db:
-            # Create the database connection only when we start to actually
-            # use it. This is essentially just a workaroud of a sqlite bug
-            # happening when 2 concurrent processes try to create the tables
-            # around the same time, the one which fails having an incorrect
-            # cache and not seeing the tables
-            self._db = sqlite3.connect(self._filename)
-            self._ensure_schema()
-        try:
-            cursor = self._db.cursor()
-            try:
-                result = method(self, cursor, *args, **kwargs)
-            finally:
-                cursor.close()
-            self._db.commit()
-        except:
-            self._db.rollback()
-            raise
-        return result
-    return inner
-
-
 class HashIdStore(object):
     """Persist package hash=>id mappings in a file.