--- a/server/sources/datafeed.py	Wed Feb 16 10:30:09 2011 +0100
+++ b/server/sources/datafeed.py	Thu Feb 17 10:24:45 2011 +0100
@@ -132,7 +132,8 @@
         self.info('pulling data for source %s', self.uri)
         for url in self.urls:
             try:
-                parser.process(url)
+                if parser.process(url):
+                    error = True
             except IOError, exc:
                 self.error('could not pull data while processing %s: %s',
                            url, exc)

--- a/sobjects/parsers.py	Wed Feb 16 10:30:09 2011 +0100
+++ b/sobjects/parsers.py	Thu Feb 17 10:24:45 2011 +0100
@@ -191,13 +191,25 @@
         # XXX suppression support according to source configuration. If set, get
         # all cwuri of entities from this source, and compare with newly
         # imported ones
+        error = False
         for item, rels in self.parse(url):
-            self.process_item(item, rels)
-            if partialcommit:
-                # commit+set_pool instead of commit(reset_pool=False) to let
-                # other a chance to get our pool
-                self._cw.commit()
-                self._cw.set_pool()
+            cwuri = item['cwuri']
+            try:
+                self.process_item(item, rels)
+                if partialcommit:
+                    # commit+set_pool instead of commit(reset_pool=False) to let
+                    # other a chance to get our pool
+                    self._cw.commit()
+                    self._cw.set_pool()
+            except ValidationError, exc:
+                if partialcommit:
+                    self.source.error('Skipping %s because of validation error %s' % (cwuri, exc))
+                    self._cw.rollback()
+                    self._cw.set_pool()
+                    error = True
+                else:
+                    raise
+        return error
 
     def parse(self, url):
         if not url.startswith('http'):