collector: preserve valid rows around ingest errors
collector: skip rejected rows after partial ingest
collector: add polling source runner